ClickHouse Exporter data loss

[Masmiiadm]

Component(s)

exporter/clickhouse

Description

I am using the OpenTelemetry Collector with the filelog receiver on a Kubernetes cluster to collect logs (see configuration below). The logs are then inserted into a ClickHouse server using the ClickHouse exporter.

However, I am noticing significant data loss. To investigate further, I limited the collection to only one container. In the collector logs, I see that 20 records were inserted:

(2024-09-28T15:38:43.890Z debug clickhouseexporter@v0.110.0/exporter_logs.go:127 insert logs {"kind": "exporter", "data_type": "logs", "name": "clickhouse", **"records": 20**, "cost": "48.506672ms"})

But when I execute the command SELECT count(*) FROM otel_logs in ClickHouse, I only see 5 records.
This means that 15 records have disappeared.

Can someone help me identify the cause of this data loss?

Steps to reproduce

Configure the OpenTelemetry Collector with the filelog receiver and ClickHouse exporter.
Limit log collection to a single container.
Check logs in the OpenTelemetry Collector and ClickHouse for discrepancies.
What is expected
The number of logs inserted into ClickHouse should match the number of records shown in the OpenTelemetry Collector logs.

What is happening

The OpenTelemetry Collector logs show that 20 records were inserted, but ClickHouse only has 5 records.

Opentelemetry config file :

exporters:
clickhouse:
 cluster_name: CLSTR01
 create_schema: false
 database: CLSTR01
 endpoint: tcp://clstr.example.local:9000
 logs_table_name: otel_logs
 metrics_table_name: otel_metrics
 username:  "user"
 password: "passwd"
 retry_on_failure:
   enabled: true
   initial_interval: 5s
   max_elapsed_time: 300s
   max_interval: 30s
 sending_queue:
   enabled: true
   queue_size: 1000
 table_engine:
   name: ReplicatedReplacingMergeTree
 timeout: 10s
 traces_table_name: otel_traces
debug: {}
extensions:
health_check:
 endpoint: ${env:MY_POD_IP}:13133
processors:
batch: {}
k8sattributes:
 extract:
   metadata:
   - k8s.namespace.name
   - k8s.pod.name
   - k8s.pod.start_time
   - k8s.pod.uid
   - k8s.deployment.name
   - k8s.node.name
   - k8s.statefulset.name
   - k8s.daemonset.name
 filter:
   node_from_env_var: K8S_NODE_NAME
 pod_association:
 - sources:
   - from: resource_attribute
     name: k8s.pod.ip
 - sources:
   - from: resource_attribute
     name: k8s.pod.uid
 - sources:
   - from: connection
memory_limiter:
 check_interval: 5s
 limit_percentage: 80
 spike_limit_percentage: 25
receivers:
filelog:
 include:
 - /var/log/pods/monitoring*/kube-state-metrics/*.log
 include_file_name: false
 include_file_path: true
 operators:
 - id: get-format
   routes:
   - expr: body matches "^\\{"
     output: parser-docker
   - expr: body matches "^[^ Z]+ "
     output: parser-crio
   - expr: body matches "^[^ Z]+Z"
     output: parser-containerd
   type: router
 - id: parser-crio
   output: extract_metadata_from_filepath
   regex: ^(?P<time>[^ Z]+) (?P<stream>stdout|stderr) (?P<logtag>[^ ]*) ?(?P<log>.*)$
   timestamp:
     layout: 2006-01-02T15:04:05.999999999Z07:00
     layout_type: gotime
     parse_from: attributes.time
   type: regex_parser
 - id: parser-containerd
   output: extract_metadata_from_filepath
   regex: ^(?P<time>[^ ^Z]+Z) (?P<stream>stdout|stderr) (?P<logtag>[^ ]*) ?(?P<log>.*)$
   timestamp:
     layout: '%Y-%m-%dT%H:%M:%S.%LZ'
     parse_from: attributes.time
   type: regex_parser
 - id: parser-docker
   output: extract_metadata_from_filepath
   timestamp:
     layout: '%Y-%m-%dT%H:%M:%S.%LZ'
     parse_from: attributes.time
   type: json_parser
 - from: attributes.log
   to: body
   type: move
 - cache:
     size: 128
   id: extract_metadata_from_filepath
   parse_from: attributes["log.file.path"]
   regex: ^.*\/(?P<namespace>[^_]+)_(?P<pod_name>[^_]+)_(?P<uid>[a-f0-9\-]{36})\/(?P<container_name>[^\._]+)\/(?P<restart_count>\d+)\.log$
   type: regex_parser
 - from: attributes.stream
   to: attributes["log.iostream"]
   type: move
 - from: attributes.container_name
   to: resource["k8s.container.name"]
   type: move
 - from: attributes.namespace
   to: resource["k8s.namespace.name"]
   type: move
 - from: attributes.pod_name
   to: resource["k8s.pod.name"]
   type: move
 - from: attributes.restart_count
   to: resource["k8s.container.restart_count"]
   type: move
 - from: attributes.uid
   to: resource["k8s.pod.uid"]
   type: move
 start_at: beginning
jaeger:
 protocols:
   grpc:
     endpoint: ${env:MY_POD_IP}:14250
   thrift_compact:
     endpoint: ${env:MY_POD_IP}:6831
   thrift_http:
     endpoint: ${env:MY_POD_IP}:14268
otlp:
 protocols:
   grpc:
     endpoint: ${env:MY_POD_IP}:4317
   http:
     endpoint: ${env:MY_POD_IP}:4318
prometheus:
 config:
   scrape_configs:
   - job_name: opentelemetry-collector
     scrape_interval: 10s
     static_configs:
     - targets:
       - ${env:MY_POD_IP}:8888
zipkin:
 endpoint: ${env:MY_POD_IP}:9411
service:
extensions:
- health_check
pipelines:
 logs:
   exporters:
   - clickhouse
   processors:
   - k8sattributes
   - resource
   receivers:
   - filelog
 metrics:
   exporters:
   - debug
   processors:
   - memory_limiter
   - batch
   receivers:
   - otlp
   - prometheus
 traces:
   exporters:
   - debug
   processors:
   - memory_limiter
   - batch
   receivers:
   - otlp
   - jaeger
   - zipkin
telemetry:
 logs:
   level: DEBUG
 metrics:
   address: ${env:MY_POD_IP}:8888
   level: detailed    ```


**clickhouse table description :** 
```yaml
1. │ Timestamp          │ DateTime64(9)                       │              │                    │         │ Delta(8), ZSTD(1) │                │
2. │ TraceId            │ String                              │              │                    │         │ ZSTD(1)           │                │
3. │ SpanId             │ String                              │              │                    │         │ ZSTD(1)           │                │
4. │ TraceFlags         │ UInt32                              │              │                    │         │ ZSTD(1)           │                │
5. │ SeverityText       │ LowCardinality(String)              │              │                    │         │ ZSTD(1)           │                │
6. │ SeverityNumber     │ Int32                               │              │                    │         │ ZSTD(1)           │                │
7. │ ServiceName        │ LowCardinality(String)              │              │                    │         │ ZSTD(1)           │                │
8. │ Body               │ String                              │              │                    │         │ ZSTD(1)           │                │
9. │ ResourceSchemaUrl  │ String                              │              │                    │         │ ZSTD(1)           │                │
10. │ ResourceAttributes │ Map(LowCardinality(String), String) │              │                    │         │ ZSTD(1)           │                │
11. │ ScopeSchemaUrl     │ String                              │              │                    │         │ ZSTD(1)           │                │
12. │ ScopeName          │ String                              │              │                    │         │ ZSTD(1)           │                │
13. │ ScopeVersion       │ String                              │              │                    │         │ ZSTD(1)           │                │
14. │ ScopeAttributes    │ Map(LowCardinality(String), String) │              │                    │         │ ZSTD(1)           │                │
15. │ LogAttributes      │ Map(LowCardinality(String), String) │              │                    │         │ ZSTD(1)  
Additional details
Any advice or insights on what might be causing this discrepancy would be greatly appreciated.

[github-actions]

Pinging code owners:

• exporter/clickhouse: @hanjm @dmitryax @Frapschen @SpencerTorres

See Adding Labels via Comments if you do not have permissions to add labels yourself.

[SpencerTorres]

I would like to see the full table DDL for otel_logs via SHOW TABLE otel_logs. I noticed you're using clustering and ReplicatedReplacingMergeTree. It's possible that the logs are being removed as duplicates.

You can also validate the rest of your configuration (memory/batch limiting) by writing logs to a file or other exporter. I see the debug log line says 20 though, so it seems like this is indeed isolated to the ClickHouse exporter/server.

You can also check the system.query_log table for the INSERT's written_rows or result_rows. I believe this value would reflect the complete count of inserted rows rather than the final rows, because replacing merge tree can still contain duplicates if parts haven't merged yet.

[SpencerTorres]

@Masmiiadm let me know if this is still an issue. As noted in the comment above I think ReplicatedReplacingMergeTree is causing similar rows to be combined, leading to the mismatch in the row counts.

[Masmiiadm]

Hello @SpencerTorres ,
Sorry for the late reply. Yes, it was indeed an issue with ReplicatedReplacingMergeTree .
Thanks for your support