WIP: Add support for logs

Note that this branch will be rebased on a periodic basis to stay on top of changes in main. If you are checking this out early, do remember to do `git pull -r` to avoid conflicts.
jjatria · Nov 27, 2023 · cfbcab0 · cfbcab0
1 parent b2b7b72
commit cfbcab0
Show file tree

Hide file tree

Showing 6 changed files with 183 additions and 38 deletions.
diff --git a/Changes b/Changes
@@ -13,6 +13,7 @@ Revision history for OpenTelemetry-Exporter-OTLP
       - otel.otlp_exporter.request_duration -> otel.exporter.otlp.request.duration
       Also: the `otel.exporter.otlp.success` metric was added for
       consistency.
+    * EXPERIMENTAL: Add support for logs
 
 0.014     2023-11-23 19:32:20+00:00 Europe/London
 

diff --git a/META.json b/META.json
@@ -28,7 +28,7 @@
          "requires" : {
             "ExtUtils::MakeMaker" : "0",
             "File::ShareDir::Install" : "0.06",
-            "perl" : "v5.20.0"
+            "perl" : "v5.32.0"
          }
       },
       "develop" : {
@@ -58,7 +58,7 @@
             "Syntax::Keyword::Dynamically" : "0",
             "Syntax::Keyword::Match" : "0",
             "Time::Piece" : "0",
-            "perl" : "v5.20.0"
+            "perl" : "v5.32.0"
          }
       },
       "test" : {
@@ -70,7 +70,7 @@
             "File::Spec" : "0",
             "Test2::V0" : "0",
             "Test::More" : "0",
-            "perl" : "v5.20.0"
+            "perl" : "v5.32.0"
          }
       }
    },

diff --git a/Makefile.PL b/Makefile.PL
@@ -2,7 +2,7 @@
 use strict;
 use warnings;
 
-use 5.020000;
+use 5.032000;
 
 use ExtUtils::MakeMaker;
 
@@ -21,7 +21,7 @@ my %WriteMakefileArgs = (
   },
   "DISTNAME" => "OpenTelemetry-Exporter-OTLP",
   "LICENSE" => "perl",
-  "MIN_PERL_VERSION" => "5.020000",
+  "MIN_PERL_VERSION" => "5.032000",
   "NAME" => "OpenTelemetry::Exporter::OTLP",
   "PREREQ_PM" => {
     "Feature::Compat::Try" => 0,

diff --git a/lib/OpenTelemetry/Exporter/OTLP.pm b/lib/OpenTelemetry/Exporter/OTLP.pm
@@ -22,6 +22,8 @@ class OpenTelemetry::Exporter::OTLP :does(OpenTelemetry::Exporter) {
     use Time::HiRes 'sleep';
     use URL::Encode 'url_decode';
 
+    use experimental 'isa';
+
     my $CAN_USE_PROTOBUF = eval {
         require Google::ProtocolBuffers::Dynamic;
         1;
@@ -70,13 +72,14 @@ class OpenTelemetry::Exporter::OTLP :does(OpenTelemetry::Exporter) {
 
     field $stopped;
     field $ua;
-    field $endpoint;
+    field $traces_endpoint;
+    field $logs_endpoint;
     field $compression :param = undef;
     field $encoder;
     field $max_retries = 5;
 
     ADJUSTPARAMS ($params) {
-        $endpoint
+        $traces_endpoint
             = delete $params->{traces_endpoint}
             // config('EXPORTER_OTLP_TRACES_ENDPOINT')
             // do {
@@ -87,6 +90,17 @@ class OpenTelemetry::Exporter::OTLP :does(OpenTelemetry::Exporter) {
                 ( $base =~ s|/+$||r ) . '/v1/traces';
             };
 
+        $logs_endpoint
+            = delete $params->{logs_endpoint}
+            // config('EXPORTER_OTLP_LOGS_ENDPOINT')
+            // do {
+                my $base = delete $params->{endpoint}
+                    // config('EXPORTER_OTLP_ENDPOINT')
+                    // 'http://localhost:4318';
+
+                ( $base =~ s|/+$||r ) . '/v1/logs';
+            };
+
         $compression
             //= config(qw( EXPORTER_OTLP_TRACES_COMPRESSION EXPORTER_OTLP_COMPRESSION ))
             // $COMPRESSION;
@@ -109,8 +123,12 @@ class OpenTelemetry::Exporter::OTLP :does(OpenTelemetry::Exporter) {
         } unless ref $headers;
 
         die OpenTelemetry::X->create(
-            Invalid => "invalid URL for OTLP exporter: $endpoint"
-        ) unless "$endpoint" =~ m|^https?://|;
+            Invalid => "invalid traces URL for OTLP exporter: $traces_endpoint"
+        ) unless "$traces_endpoint" =~ m|^https?://|;
+
+        die OpenTelemetry::X->create(
+            Invalid => "invalid logs URL for OTLP exporter: $logs_endpoint"
+        ) unless "$logs_endpoint" =~ m|^https?://|;
 
         die OpenTelemetry::X->create(
             Unsupported => "unsupported compression key for OTLP exporter: $compression"
@@ -200,7 +218,7 @@ class OpenTelemetry::Exporter::OTLP :does(OpenTelemetry::Exporter) {
         return 1;
     }
 
-    method $send_request ( $data, $timeout ) {
+    method $send_request ( $endpoint, $data, $timeout ) {
         my %request = ( content => $data );
 
         $metrics->report_distribution(
@@ -274,7 +292,8 @@ class OpenTelemetry::Exporter::OTLP :does(OpenTelemetry::Exporter) {
                                 );
 
                                 OpenTelemetry->handle_error(
-                                    message => "Unhandled error sending OTLP request: $res->{content}",
+                                    exception => $res->{content},
+                                    message   => 'Unhandled error sending OTLP request',
                                 );
 
                                 return TRACE_EXPORT_FAILURE;
@@ -290,21 +309,27 @@ class OpenTelemetry::Exporter::OTLP :does(OpenTelemetry::Exporter) {
                     );
 
                     if ( $CAN_USE_PROTOBUF ) {
-                        require OpenTelemetry::Proto;
+                        my %error;
+
                         try {
+                            require OpenTelemetry::Proto;
+
                             my $status = OTel::Google::RPC::Status
                                 ->decode($res->{content});
-                            OpenTelemetry->handle_error(
-                                exception => 'OTLP exporter received an RPC error status',
-                                message   => $status->encode_json,
+
+                            %error = (
+                                exception => $status->encode_json,
+                                message   => 'OTLP exporter received an RPC error status',
                             );
                         }
                         catch($e) {
-                            OpenTelemetry->handle_error(
+                            %error = (
                                 exception => $e,
                                 message   => 'Unexpected error decoding RPC status in OTLP exporter',
                             );
                         }
+
+                        OpenTelemetry->handle_error(%error);
                     }
 
                     my $after = $res->{status} =~ /^(?: 429 | 503 )$/x
@@ -325,14 +350,36 @@ class OpenTelemetry::Exporter::OTLP :does(OpenTelemetry::Exporter) {
         }
     }
 
-    method export ( $spans, $timeout = undef ) {
+    method export ( $data, $timeout = undef ) {
         return TRACE_EXPORT_FAILURE if $stopped;
+        return unless @$data;
+
+        try {
+            dynamically OpenTelemetry::Context->current
+                = OpenTelemetry::Trace->untraced_context;
+
+            my $request = $encoder->encode($data);
+
+            my $endpoint;
+
+            $endpoint //= $logs_endpoint
+                if $data->[0] isa OpenTelemetry::SDK::Logs::LogRecord;
+
+            $endpoint //= $traces_endpoint;
 
-        dynamically OpenTelemetry::Context->current
-            = OpenTelemetry::Trace->untraced_context;
+            my $result = $self->$send_request( $endpoint, $request, $timeout );
 
-        my $request = $encoder->encode($spans);
-        $self->$send_request( $request, $timeout );
+            $metrics->inc_counter('otel.exporter.otlp.success');
+            $metrics->inc_counter_by(
+                'otel.exporter.otlp.exported' => scalar @$data,
+            );
+
+            return $result;
+        }
+        catch($e) {
+            warn "Could not export data: $e";
+            return TRACE_EXPORT_FAILURE;
+        }
     }
 
     async method shutdown ( $timeout = undef ) {

diff --git a/lib/OpenTelemetry/Exporter/OTLP/Encoder/JSON.pm b/lib/OpenTelemetry/Exporter/OTLP/Encoder/JSON.pm
@@ -11,6 +11,8 @@ class OpenTelemetry::Exporter::OTLP::Encoder::JSON {
     use Ref::Util qw( is_hashref is_arrayref );
     use Scalar::Util 'refaddr';
 
+    use experimental 'isa';
+
     method content_type () { 'application/json' }
 
     method serialise ($data) { encode_json $data }
@@ -53,7 +55,7 @@ class OpenTelemetry::Exporter::OTLP::Encoder::JSON {
         };
     }
 
-    method encode_event ( $event ) {
+    method encode_span_event ( $event ) {
         {
             attributes             => $self->encode_kvlist($event->attributes),
             droppedAttributesCount => $event->dropped_attributes,
@@ -62,7 +64,7 @@ class OpenTelemetry::Exporter::OTLP::Encoder::JSON {
         };
     }
 
-    method encode_link ( $link ) {
+    method encode_span_link ( $link ) {
         {
             attributes             => $self->encode_kvlist($link->attributes),
             droppedAttributesCount => $link->dropped_attributes,
@@ -71,7 +73,7 @@ class OpenTelemetry::Exporter::OTLP::Encoder::JSON {
         };
     }
 
-    method encode_status ( $status ) {
+    method encode_span_status ( $status ) {
         {
             code    => $status->code,
             message => $status->description,
@@ -85,13 +87,13 @@ class OpenTelemetry::Exporter::OTLP::Encoder::JSON {
             droppedEventsCount     => $span->dropped_events,
             droppedLinksCount      => $span->dropped_links,
             endTimeUnixNano        => int $span->end_timestamp * 1_000_000_000,
-            events                 => [ map $self->encode_event($_), $span->events ],
+            events                 => [ map $self->encode_span_event($_), $span->events ],
             kind                   => $span->kind,
-            links                  => [ map $self->encode_link($_),  $span->links  ],
+            links                  => [ map $self->encode_span_link($_),  $span->links  ],
             name                   => $span->name,
             spanId                 => $span->hex_span_id,
             startTimeUnixNano      => int $span->start_timestamp * 1_000_000_000,
-            status                 => $self->encode_status($span->status),
+            status                 => $self->encode_span_status($span->status),
             traceId                => $span->hex_trace_id,
             traceState             => $span->trace_state->to_string,
         };
@@ -103,6 +105,21 @@ class OpenTelemetry::Exporter::OTLP::Encoder::JSON {
         $data;
     }
 
+    method encode_log ( $log ) {
+        {
+            attributes             => $self->encode_attributes($log->attributes),
+            droppedAttributesCount => $log->dropped_attributes,
+            timeUnixNano           => int $log->timestamp * 1_000_000_000,
+            observedTimeUnixNano   => int $log->observed_timestamp * 1_000_000_000,
+            severityText           => $log->severity_text,
+            severityNumber         => $log->severity_number,
+            body                   => $self->encode_anyvalue( $log->body ),
+            flags                  => $log->trace_flags->flags,
+            spanId                 => $log->hex_span_id,
+            traceId                => $log->hex_trace_id,
+        };
+    }
+
     method encode_scope ( $scope ) {
         {
             attributes             => $self->encode_kvlist($scope->attributes),
@@ -119,6 +136,13 @@ class OpenTelemetry::Exporter::OTLP::Encoder::JSON {
         };
     }
 
+    method encode_scope_logs ( $scope, $logs ) {
+        {
+            scope      => $self->encode_scope($scope),
+            logRecords => [ map $self->encode_log($_), @$logs ],
+        };
+    }
+
     method encode_resource_spans ( $resource, $spans ) {
         my %scopes;
 
@@ -138,18 +162,43 @@ class OpenTelemetry::Exporter::OTLP::Encoder::JSON {
         };
     }
 
-    method encode ( $spans ) {
+    method encode_resource_logs ( $resource, $logs ) {
+        my %scopes;
+
+        for (@$logs) {
+            my $key = refaddr $_->instrumentation_scope;
+
+            $scopes{ $key } //= [ $_->instrumentation_scope, [] ];
+            push @{ $scopes{ $key }[1] }, $_;
+        }
+
+        {
+            resource => $self->encode_resource($resource),
+            scopeLogs => [
+                map $self->encode_scope_logs(@$_), values %scopes,
+            ],
+            schemaUrl => $resource->schema_url,
+        };
+    }
+
+    method encode ( $data ) {
         my ( %request, %resources );
 
-        for (@$spans) {
+        my $type;
+        for (@$data) {
+            $type //= $_ isa OpenTelemetry::SDK::Logs::LogRecord
+                ? 'logs'
+                : 'spans';
+
             my $key = refaddr $_->resource;
             $resources{ $key } //= [ $_->resource, [] ];
             push @{ $resources{ $key }[1] }, $_;
         }
 
+        my $encode = "encode_resource_$type";
         $self->serialise({
-            resourceSpans => [
-                map $self->encode_resource_spans(@$_), values %resources,
+            'resource' . ucfirst $type => [
+                map $self->$encode(@$_), values %resources,
             ]
         });
     }