tracing: add span on stream complete (#10)

server/server.go

@@ -59,6 +59,12 @@ func WithFilters(f ...filter.Filter) Option {
 	}
 }
 
+func WithServiceOptions(opts ...service.Option) Option {
+	return func(s *Server) {
+		s.serviceOpts = append(s.serviceOpts, opts...)
+	}
+}
+
 func WithGrpcServer(server *grpc.Server, network string, address string) Option {
 	return func(s *Server) {
 		s.grpcServer = server

service/service.go

@@ -24,13 +24,13 @@ const (
 )
 
 var (
-	ProcessResourceName          = "Process"
 	RequestHeadersResourceName   = "RequestHeaders"
 	RequestBodyResourceName      = "RequestBody"
 	RequestTrailersResourceName  = "RequestTrailers"
 	ResponseHeadersResourceName  = "ResponseHeaders"
 	ResponseBodyResourceName     = "ResponseBody"
 	ResponseTrailersResourceName = "ResponseTrailers"
+	StreamCompleteResourceName   = "StreamComplete"
 )
 
 type ExtProcessor struct {
@@ -62,10 +62,15 @@ func (svc *ExtProcessor) Process(procsrv extproc.ExternalProcessor_ProcessServer
 	ctx := procsrv.Context()
 	if len(svc.streamCallbacks) > 0 {
 		defer func() {
+			ctx, span := svc.tracer.Start(ctx, StreamCompleteResourceName)
+			defer span.End()
 			for _, s := range svc.streamCallbacks {
+				resourceName := fmt.Sprintf("%T/%s", s, StreamCompleteResourceName)
+				_, span := svc.tracer.Start(ctx, resourceName)
 				if err := s.OnStreamComplete(req); err != nil {
-					slog.Error(fmt.Sprintf("%T.OnStreamComplete returned an error", s), "err", err.Error())
+					slog.Error(fmt.Sprintf("%T.%s returned an error", s, StreamCompleteResourceName), "err", err.Error())
 				}
+				span.End()
 			}
 		}()
 	}

test/containers/envoy/envoy.yml

@@ -105,23 +105,9 @@ static_resources:
                       address: host.testcontainers.internal
                       port_value: 8080
 
-    - name: datadog
-      connect_timeout: 1s
-      type: STRICT_DNS
-      lb_policy: round_robin
-      load_assignment:
-        cluster_name: datadog
-        endpoints:
-          - lb_endpoints:
-              - endpoint:
-                  address:
-                    socket_address:
-                      address: localhost
-                      port_value: 8126
-
 overload_manager:
   resource_monitors:
-    - name: "envoy.resource_monitors.global_downstream_max_connections"
+    - name: envoy.resource_monitors.global_downstream_max_connections
       typed_config:
         "@type": type.googleapis.com/envoy.extensions.resource_monitors.downstream_connections.v3.DownstreamConnectionsConfig
         max_active_downstream_connections: 1000