feat(plugin-aws-sdk): add span kind to request hook

Author: Amir Blum

packages/plugin-aws-sdk/src/aws-sdk.ts

@@ -9,7 +9,7 @@
     callback    |       1           |       2   
  */
 import { BasePlugin } from "@opentelemetry/core";
-import { Span, CanonicalCode, Attributes } from "@opentelemetry/api";
+import { Span, CanonicalCode, Attributes, SpanKind } from "@opentelemetry/api";
 import * as shimmer from "shimmer";
 import AWS from "aws-sdk";
 import { AttributeNames } from "./enums";
@@ -71,12 +71,14 @@ class AwsPlugin extends BasePlugin<typeof AWS> {
 
   private _startAwsSpan(
     request: AWS.Request<any, any>,
-    additionalAttributes?: Attributes
+    additionalAttributes?: Attributes,
+    spanKind?: SpanKind
   ): Span {
     const operation = (request as any).operation;
     const service = (request as any).service;
 
     const newSpan = this._tracer.startSpan(this._getSpanName(request), {
+      kind: spanKind,
       attributes: {
         [AttributeNames.COMPONENT]: this.moduleName,
         [AttributeNames.AWS_OPERATION]: operation,
@@ -147,7 +149,8 @@ class AwsPlugin extends BasePlugin<typeof AWS> {
       );
       const span = thisPlugin._startAwsSpan(
         awsRequest,
-        requestMetadata.spanAttributes
+        requestMetadata.spanAttributes,
+        requestMetadata.spanKind
       );
       thisPlugin._callPreRequestHooks(span, awsRequest);
       thisPlugin._registerCompletedEvent(span, awsRequest);
@@ -179,7 +182,8 @@ class AwsPlugin extends BasePlugin<typeof AWS> {
       );
       const span = thisPlugin._startAwsSpan(
         awsRequest,
-        requestMetadata.spanAttributes
+        requestMetadata.spanAttributes,
+        requestMetadata.spanKind
       );
       thisPlugin._callPreRequestHooks(span, awsRequest);
       thisPlugin._registerCompletedEvent(span, awsRequest);

packages/plugin-aws-sdk/src/services/ServiceExtension.ts

@@ -1,15 +1,16 @@
-import { Span, Attributes } from "@opentelemetry/api";
+import { Span, Attributes, SpanKind } from "@opentelemetry/api";
 
 /*
 isIncoming - if true, then the operation callback / promise should be bind with the operation's span 
 */
 export interface RequestMetadata {
   isIncoming: boolean;
   spanAttributes?: Attributes;
+  spanKind?: SpanKind;
 }
 
 export interface ServiceExtension {
-  // called before request is sent, and before span is created
+  // called before request is sent, and before span is started
   requestHook: (request: AWS.Request<any, any>) => RequestMetadata;
   responseHook: (response: AWS.Response<any, any>, span: Span) => void;
 }

packages/plugin-aws-sdk/src/services/sqs.ts

@@ -30,6 +30,7 @@ export class SqsServiceExtension implements ServiceExtension {
   requestHook(request: AWS.Request<any, any>): RequestMetadata {
     const queueUrl = this.extractQueueUrl(request);
     const queueName = this.extractQueueNameFromUrl(queueUrl);
+    let spanKind: SpanKind = SpanKind.CLIENT;
 
     const spanAttributes = {
       [SqsAttributeNames.MESSAGING_SYSTEM]: "aws.sqs",
@@ -44,18 +45,25 @@ export class SqsServiceExtension implements ServiceExtension {
     switch (operation) {
       case "receiveMessage":
         isIncoming = true;
+        spanKind = SpanKind.CONSUMER;
         spanAttributes[SqsAttributeNames.MESSAGING_OPERATION] = "receive";
         break;
+
+      case "sendMessage":
+      case "sendMessageBatch":
+        spanKind = SpanKind.PRODUCER;
+        break;
     }
 
     return {
       isIncoming,
       spanAttributes,
+      spanKind,
     };
   }
 
   responseHook = (response: AWS.Response<any, any>, span: Span) => {
-    const messages: AWS.SQS.Message[] = response.data.Messages;
+    const messages: AWS.SQS.Message[] = response?.data?.Messages;
     if (messages) {
       const queueUrl = this.extractQueueUrl((response as any)?.request);
       const queueName = this.extractQueueNameFromUrl(queueUrl);