Merge pull request #2352 from flow-php/flow-telemetry-improvements

Flow telemetry improvements

Author: norberttech

.nix/pkgs/flow-php/package.nix

@@ -11,7 +11,8 @@
     with-blackfire ? false,
     with-pg-query-ext ? false,
     with-arrow-ext ? false,
-    with-grpc ? false
+    with-grpc ? false,
+    with-protobuf ? true
 }:
 
 let
@@ -27,7 +28,6 @@ let
             (php-lz4.override { inherit php; })
             (php-snappy.override { inherit php; })
             (php-zstd.override { inherit php; })
-            protobuf
             xmlreader
             xmlwriter
             zlib
@@ -38,6 +38,7 @@ let
         ++ (if with-pg-query-ext then [(php-pg-query-ext.override { inherit php; })] else [])
         ++ (if with-arrow-ext then [(php-arrow-ext.override { inherit php; })] else [])
         ++ (if with-grpc then [grpc] else [])
+        ++ (if with-protobuf then [protobuf] else [])
     );
 in
 flowPHP.buildEnv {

.php-cs-fixer.php

@@ -33,6 +33,8 @@
     ])
     ->exclude([
         'Flow/Parquet/ThriftModel',
+        'Opentelemetry',
+        'GPBMetadata',
         'Flow/CLI/Tests/Integration',
         'Flow/ETL/Tests/Unit/Loader',
         'Flow/ETL/Tests/Unit/Exception',

compose.yml.dist

@@ -107,7 +107,7 @@ services:
             - "8888:8888"
         volumes:
             - "./docker/otel-collector-config.yaml:/etc/otel-collector-config.yaml:ro"
-            - "./var/otel:/var/otel"
+            - "./.scratchpad/migraitons/var/otel:/var/otel"
         depends_on:
             - aspire
             - postgres

composer.json

@@ -58,7 +58,6 @@
         "grpc/grpc": "^1.74",
         "nikic/php-parser": "^5.3",
         "nyholm/psr7": "^1.8",
-        "open-telemetry/gen-otlp-protobuf": "^1.8",
         "php-http/curl-client": "^2.2",
         "php-http/mock-client": "^1.5",
         "ramsey/uuid": "^4.5",
@@ -187,6 +186,12 @@
             ],
             "Flow\\ETL\\Adapter\\Doctrine\\": [
                 "src/adapter/etl-adapter-doctrine/src/Flow/ETL/Adapter/Doctrine"
+            ],
+            "Opentelemetry\\Proto\\": [
+                "src/bridge/telemetry/otlp/src/Opentelemetry/Proto"
+            ],
+            "GPBMetadata\\Opentelemetry\\": [
+                "src/bridge/telemetry/otlp/src/GPBMetadata/Opentelemetry"
             ]
         },
         "files": [
@@ -428,6 +433,9 @@
             "protoc --php_out=src/lib/postgresql/src --proto_path=src/lib/postgresql/resources/proto pg_query.proto",
             "@cs:php:fix"
         ],
+        "build:telemetry:otlp:protobuf": [
+            "src/bridge/telemetry/otlp/bin/generate-proto.sh"
+        ],
         "pre-autoload-dump": [
             "Google\\Task\\Composer::cleanup"
         ],

composer.lock

@@ -6,6 +6,13 @@ receivers:
       http:
         endpoint: 0.0.0.0:4318
 
+  # OTLP JSON file receiver — picks up JSONL written by flow-php StreamTransport.
+  # Docs: https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/receiver/otlpjsonfilereceiver
+  otlpjsonfile:
+    include:
+      - /var/otel/*.jsonl
+    start_at: end
+
   # PostgreSQL metrics receiver
   # Docs: https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/receiver/postgresqlreceiver
   postgresql:
@@ -45,6 +52,16 @@ exporters:
     endpoint: aspire:18889
     tls:
       insecure: true
+    sending_queue:
+      enabled: true
+      storage: file_storage/queue
+      num_consumers: 4
+      queue_size: 10000
+    retry_on_failure:
+      enabled: true
+      initial_interval: 5s
+      max_interval: 30s
+      max_elapsed_time: 0
 
 processors:
   # Add service.name to PostgreSQL metrics
@@ -80,9 +97,15 @@ processors:
 extensions:
   health_check:
     endpoint: 0.0.0.0:13133
+  file_storage/queue:
+    directory: /var/lib/otelcol/queue
+    timeout: 10s
+    compaction:
+      directory: /var/lib/otelcol/queue
+      on_start: true
 
 service:
-  extensions: [health_check]
+  extensions: [health_check, file_storage/queue]
   telemetry:
     traces:
       sampler:
@@ -138,3 +161,6 @@ service:
     logs:
       receivers: [otlp]
       exporters: [debug, otlp_grpc/aspire]
+    logs/file:
+      receivers: [otlpjsonfile]
+      exporters: [debug, otlp_grpc/aspire]

docker/otel-collector-config.yaml

@@ -286,7 +286,7 @@ use function Flow\Telemetry\DSL\{
 use function Flow\Bridge\Telemetry\OTLP\DSL\{
     otlp_curl_transport,
     otlp_json_serializer,
-    otlp_log_exporter,
+    otlp_exporter,
 };
 use function Flow\Bridge\Monolog\Telemetry\DSL\telemetry_handler;
 
@@ -305,7 +305,7 @@ $transport = otlp_curl_transport(
 $telemetry = telemetry(
     $resource,
     loggerProvider: logger_provider(
-        batching_log_processor(otlp_log_exporter($transport))
+        batching_log_processor(otlp_exporter($transport))
     ),
 );