[ClickHouse org] compiles with logstransformprocessor

cmd/otelcontribcol/builder-config.yaml

@@ -93,6 +93,7 @@ processors:
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/k8sattributesprocessor v0.89.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricsgenerationprocessor v0.89.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstransformprocessor v0.89.0
+  - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/logstransformprocessor v0.89.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/probabilisticsamplerprocessor v0.89.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/redactionprocessor v0.89.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/resourcedetectionprocessor v0.89.0
@@ -324,6 +325,7 @@ replaces:
   - github.com/open-telemetry/opentelemetry-collector-contrib/receiver/influxdbreceiver => ../../receiver/influxdbreceiver
   - github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awscontainerinsightreceiver => ../../receiver/awscontainerinsightreceiver
   - github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstransformprocessor => ../../processor/metricstransformprocessor
+  - github.com/open-telemetry/opentelemetry-collector-contrib/processor/logstransformprocessor => ../../processor/logstransformprocessor
   - github.com/open-telemetry/opentelemetry-collector-contrib/extension/sigv4authextension => ../../extension/sigv4authextension
   - github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/opencensus => ../../pkg/translator/opencensus
   - github.com/open-telemetry/opentelemetry-collector-contrib/exporter/pulsarexporter => ../../exporter/pulsarexporter

cmd/otelcontribcol/go.mod

@@ -86,6 +86,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/k8sattributesprocessor v0.89.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricsgenerationprocessor v0.89.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstransformprocessor v0.89.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/processor/logstransformprocessor v0.89.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/probabilisticsamplerprocessor v0.89.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/redactionprocessor v0.89.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/remotetapprocessor v0.89.0
@@ -890,6 +891,8 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8scl
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor => ../../processor/transformprocessor
 
+replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/logstransformprocessor => ../../processor/logstransformprocessor
+
 replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/filterprocessor => ../../processor/filterprocessor
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/extension/observer/hostobserver => ../../extension/observer/hostobserver
@@ -956,6 +959,8 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsco
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstransformprocessor => ../../processor/metricstransformprocessor
 
+replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/logstransformprocessor => ../../processor/logstransformprocessor
+
 replace github.com/open-telemetry/opentelemetry-collector-contrib/extension/sigv4authextension => ../../extension/sigv4authextension
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/opencensus => ../../pkg/translator/opencensus

exporter/clickhouseexporter/example/docker-compose.yml

@@ -4,18 +4,28 @@ networks:
   otel-clickhouse:
 
 services:
+#  clickhouse-logger:
+#    image: alpine:latest
+#    container_name: alpine
+#    entrypoint: ["head", "-n", "1000", "/tmp/clickhouse-server.log"]
+#    volumes:
+#      - "./clickhouse-server.log:/tmp/clickhouse-server.log"
+
   otelcollector:
     build:
       context: ../../../local
-      dockerfile: ../exporter/clickhouseexporter/example/Dockerfile
+      dockerfile: Dockerfile
     # Uncomment the next line to use a preexisting image
-    # image: otelcontribcol:latest
+    image: otelcontribcol:latest
     container_name: otel
     command:
       - "--config=/etc/otel-collector-config.yml"
-      - "--set=service.telemetry.logs.level=INFO"
+      - "--set=service.telemetry.logs.level=DEBUG"
     volumes:
       - ./otel-collector-config.yml:/etc/otel-collector-config.yml
+      - ./clickhouse-server.log:/tmp/clickhouse-server.log
+      - ./clickhouse-server.log.short:/tmp/clickhouse-server.log.short
+      - ./otel-storage:/var/lib/otelcol
     ports:
       - "4317:4317"   # otlp receiver
       - "1888:1888"   # pprof extension
@@ -49,35 +59,36 @@ services:
     networks:
       - otel-clickhouse
 
-  # Log generator
-  flog:
-    image: mingrammer/flog:0.4.3
-    # Output fake log in JSON format
-    command: [ "--format=json", "--loop", "--delay=100ms"]
-    networks:
-      - otel-clickhouse
-    depends_on:
-      - otelcollector
-    logging:
-      driver: fluentd
-      options:
-        mode: non-blocking
-        # Allow time for otel-collector to spin up, then forward fluentd logs to the fluentforwarder receiver.
-        fluentd-async-connect: "true"
-        # Use nanosecond precision
-        fluentd-sub-second-precision: "true"
-    stop_signal: SIGKILL
+#  # Log generator
+#  flog:
+#    image: mingrammer/flog:0.4.3
+#    # Output fake log in JSON format
+#    command: [ "--format=json", "--loop", "--delay=100ms"]
+#    networks:
+#      - otel-clickhouse
+#    depends_on:
+#      - otelcollector
+#    logging:
+#      driver: fluentd
+#      options:
+#        mode: non-blocking
+#        # Allow time for otel-collector to spin up, then forward fluentd logs to the fluentforwarder receiver.
+#        fluentd-async-connect: "true"
+#        # Use nanosecond precision
+#        fluentd-sub-second-precision: "true"
+#    stop_signal: SIGKILL
 
-  # Traces generator
-  tracegen:
-    build:
-      context: ../../../cmd/tracegen/
-    command:
-      - --otlp-endpoint=otelcollector:4317
-      - --otlp-insecure
-      - --rate=100
-      - --duration=10000h
-    networks:
-      - otel-clickhouse
-    depends_on:
-      - otelcollector
+
+#  # Traces generator
+#  tracegen:
+#    build:
+#      context: ../../../cmd/tracegen/
+#    command:
+#      - --otlp-endpoint=otelcollector:4317
+#      - --otlp-insecure
+#      - --rate=100
+#      - --duration=10000h
+#    networks:
+#      - otel-clickhouse
+#    depends_on:
+#      - otelcollector

exporter/clickhouseexporter/example/otel-collector-config.yml

@@ -1,34 +1,130 @@
 receivers:
-  fluentforward:
-    endpoint: 0.0.0.0:24224
-  otlp:
-    protocols:
-      grpc:
-        endpoint: 0.0.0.0:4317
+  filelog:
+    start_at: beginning
+#    storage: file_storage
+    include:
+      - /tmp/clickhouse-server.log.short
+#    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.000000000-07: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
+#      - id: extract_metadata_from_filepath
+#        parse_from: attributes["log.file.path"]
+#        regex: ^.*\/(?P<namespace>[^_]+)_(?P<pod_name>[^_]+)_(?P<uid>[a-f0-9\-]+)\/(?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
+#      - from: attributes.log
+#        to: body
+#        type: move
+
+
+# Error code regex: [A-Z][A-Z0-9]+_[A-Z0-9_]+
 processors:
+#  logstransform:
+#    operators:
+#      - type: regex_parser
+#        regex: '(?<exception>[A-Z][A-Z0-9]+_[A-Z0-9_]+)'
+#        clickhouse_exception:
+#          parse_from:
+#            body.exception
+  transform:
+    log_statements:
+      - context: resource
+        statements:
+          - set(attributes["k8s.container.name"], "c-teal-x-y-43-server")
+          - set(attributes["app"], "") where attributes["app"] == nil
+          - set(attributes["app"], "clickhouse-server") where IsMatch(attributes["k8s.container.name"], "^c-.+-server$")
+          - set(attributes["app"], "clickhouse-keeper") where IsMatch(attributes["k8s.container.name"], "^c-.+-keeper$")
+      - context: log
+        statements:
+          - merge_maps(attributes, ExtractPatterns(body, "(?P<exception>[A-Z][A-Z0-9]+_[A-Z0-9_]+)"), "insert")
+          - set(severity_text, "TRACE") where IsMatch(body, "[<]Trace[>]") == true
+          - set(severity_text, "DEBUG") where IsMatch(body, "[<]Debug[>]") == true
+          - set(severity_text, "INFO") where IsMatch(body, "[<]Info(rmation(al)?)?[>]")
+            == true
+          - set(severity_text, "WARN") where IsMatch(body, "[<]Warn(ing)?[>]") == true
+          - set(severity_text, "ERROR") where IsMatch(body, "[<]Error[>]") == true
+          - set(severity_text, "FATAL") where IsMatch(body, "[<]Fatal[>]") == true
+          - merge_maps(attributes, ParseJSON(body), "insert") where IsMatch(body, "^\\s*[{].+?[}]\\s*$")
+            == true
+          - set(body, attributes["msg"]) where attributes["msg"] != nil
+          - set(body, attributes["message"]) where attributes["message"] != nil
+          - delete_key(attributes, "msg") where body == attributes["msg"]
+          - delete_key(attributes, "message") where body == attributes["message"]
+          - set(severity_text, ConvertCase(attributes["level"], "upper")) where IsMatch(attributes["level"],
+            "(?i)^(trace|debug|info|warn|error|fatal)$") == true
+          - delete_key(attributes, "level") where attributes["level"] != nil and severity_text
+            == ConvertCase(attributes["level"], "upper")
   batch:
-    send_batch_size: 100000
-    timeout: 5s
-  memory_limiter:
-    check_interval: 2s
-    limit_mib: 1800
-    spike_limit_mib: 500
-  resourcedetection/system:
-    detectors: [ "system" ]
-    system:
-      hostname_sources: [ "os" ]
-  resource:
-    attributes:
-      - key: service.name
-        value: "serviceName"
-        action: upsert
+    send_batch_size: 10
+    timeout: 2s
+#  memory_limiter:
+#    check_interval: 2s
+#    limit_mib: 1800
+#    spike_limit_mib: 500
+#  resourcedetection/system:
+#    detectors: [ "system" ]
+#    system:
+#      hostname_sources: [ "os" ]
+#  resource:
+#    attributes:
+#      - key: service.name
+#        value: "serviceName"
+#        action: upsert
 exporters:
   clickhouse:
-    endpoint: tcp://clickhouse:9000
+    endpoint: http://clickhouse:8123
     database: otel
-    logs_table_name: otel_logs
+    logs_table_name: otel_logs_1
     traces_table_name: otel_traces
-    ttl: 12h
+    connection_params:
+      async_insert: 1
+      wait_for_async_insert: 0
+    ttl_days: 12h
     timeout: 10s
     sending_queue:
       queue_size: 100
@@ -38,20 +134,22 @@ exporters:
       max_interval: 30s
       max_elapsed_time: 300s
 extensions:
+#  file_storage:
+#    directory: /var/lib/otelcol
   health_check:
   pprof:
   zpages:
   memory_ballast:
     size_mib: 1000
 
 service:
-  extensions: [ pprof, zpages, health_check ]
+  extensions:
+    - pprof
+    - zpages
+    - health_check
+#    - file_storage
   pipelines:
     logs:
-      receivers: [ fluentforward, otlp ]
-      processors: [ memory_limiter, resourcedetection/system, resource, batch ]
-      exporters: [ clickhouse ]
-    traces:
-      receivers: [ otlp ]
-      processors: [ memory_limiter, resourcedetection/system, resource, batch ]
+      receivers: [  filelog ]
+      processors: [  transform, batch ]
       exporters: [ clickhouse ]