New Relic の OpenTelemetry トレース

OpenTelemetry 、トレーステレメトリを記録するためのAPI 、スパンデータをエクスポートするためのSDK 、アプリケーション境界を越えてトレースするためのコンテキストトレースなど、豊富なトレースエコシステムを提供します。

このページでは、New Relic が New Relic OTLP エンドポイント経由で受信した OpenTelemetry スパンをどのように処理するかについて説明します。 OpenTelemetryをNew Relicに送信するには、OTLP エンドポイント経由でNew Relic OpenTelemetryにデータをエクスポートするように Trace ソースを設定する必要があります。次のページを参照してください。

エンドポイント設定の要件については、 New Relic OTLP エンドポイントを参照してください。
OpenTelemetryを使用してサービスを構成する手順については、 OpenTelemetry APM監視」を参照してください。

OTLPスパンマッピング

New Relic は OTLP スパンをSpanデータ型にマッピングします。次の表は、トレースプロトコルメッセージ定義のフィールドがどのように解釈されるかを示しています。

OTLPプロトフィールド	New Relic `Span`フィールド
`ResourceSpans.Resource.attributes`	各キー値は`Span`の属性である[1]
`ScopeSpans.InstrumentationScope.name`	`otel.library.name`
`ScopeSpans.InstrumentationScope.version`	`otel.library.version`
`ScopeSpans.InstrumentationScope.attributes`	各キー値は`Span`の属性である[1]
`Span.trace_id`	`trace.id`
`Span.span_id`	`id`
`Span.trace_state`	`w3c.tracestate`
`Span.parent_span_id`	`parent.id`
`Span.name`	`name`
`Span.kind`	`span.kind`
`Span.start_time_unix_nano`	`timestamp`
`Span.end_time_unix_nano`	`duration.ms` ( `Span.start_time_unix_nano`で計算)
`Span.attributes`	各キー値は`Span`の属性である[1]
`Span.dropped_attribute_count`	`otel.dropped_attributes_count`
`Span.events`	各イベントは`SpanEvent`として記録され、 `span.id` / `trace.id`はソーススパンを参照し、カウントは次のように保存されます。 `nr.spanEventCount`
`Span.events[*].time_unix_nano`	`timestamp`として保存 `SpanEvent`
`Span.events[*].name`	`name`として保存 `SpanEvent`
`Span.events[*].attributes`	各キー値は、 `SpanEvent`
`Span.events[*].dropped_attributes_count`	`ote.dropped_Attributes_count`として保存 `SpanEvent`
`Span.dropped_events_count`	`otel.dropped_events_count`
`Span.status.message`	`otel.status_description`
`Span.status.code`	`otel.status_code`

表の脚注

[1]リソース属性、スコープ属性、スパン属性、および最上位のスパンフィールドで競合が発生した場合、優先順位（最高から最低）は、最上位のSpan.*フィールド> Span.attributes > ScopeSpans.InstrumentationScope.attributes > ResourceSpans.Resource.attributesとなります。

OTLP New Relicエンドポイントでサポートされるプロパティタイプの詳細については「OTLP プロパティタイプ」を、プロパティに対して実行される検証の詳細については「OTLP プロパティ制限」を参照してください。

スパンのリンク

New Relic は OpenTelemetryスパンリンクをサポートしており、これにより直接の親子関係のないスパン間の因果関係を作成できます。スパンリンクは、メッセージキュー、イベントストリーム、バッチ処理システムなどの非同期境界を越えて分割されるディストリビューティッド（分散）トレーシングを理解するために不可欠です。

スパンリンクを使用する場合

次のシナリオで span リンクを使用します。

Message queue producers and consumers: AWS SQS、RabbitMQ、Kafka などのキューからのメッセージを処理するときに、消費スパンを生成スパンにリンクします。
Fan-in patterns: 複数のプロデューサートレースを、その出力を集約する単一のコンシューマートレースにリンクします。
Batch processing: バッチ処理されたメッセージを個々の元のトレースに返すリンクスパン。
Long-running workflows: 通常のトレース期間の制限を超えるワークフローステップにまたがる接続。

スパンリンクの実装

OpenTelemetryメーターにスパンリンクを実装するには、次の手順を実行する必要があります。

受信メッセージまたはイベントからトレースコンテキストを抽出する
消費者で新しいスパンを開始するときにスパンリンクを作成します。
トレースコンテキストがメッセージングインフラストラクチャを通じて伝播されることを確認する

次の例は、さまざまな言語で span リンクを実装する方法を示しています。

from opentelemetry import trace
from opentelemetry.trace import Link, SpanContext, TraceFlags
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator

tracer = trace.get_tracer(__name__)
propagator = TraceContextTextMapPropagator()

# Producer: Publishing a message with trace context
def publish_message(queue, message_body):
    with tracer.start_as_current_span("publish_message") as span:
        # Prepare message with trace context headers
        carrier = {}
        propagator.inject(carrier)

        # Add carrier headers to your message metadata
        message = {
            'body': message_body,
            'headers': carrier
        }

        queue.publish(message)
        span.set_attribute("messaging.destination", queue.name)
        span.set_attribute("messaging.system", "custom_queue")

# Consumer: Processing a message with span link
def process_message(message):
    # Extract trace context from message headers
    carrier = message.get('headers', {})
    ctx = propagator.extract(carrier)

    # Get the span context from the extracted context
    span_context = trace.get_current_span(ctx).get_span_context()

    # Create a new span with a link to the producer span
    links = []
    if span_context.is_valid:
        links = [Link(span_context)]

    with tracer.start_as_current_span(
        "process_message",
        links=links
    ) as span:
        # Process the message
        result = handle_message(message['body'])

        span.set_attribute("messaging.system", "custom_queue")
        span.set_attribute("messaging.operation", "process")

        return result

AWS SQS の場合:

import boto3
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator

sqs = boto3.client('sqs')
propagator = TraceContextTextMapPropagator()

# Publishing to SQS
def send_sqs_message(queue_url, message_body):
    with tracer.start_as_current_span("sqs_publish") as span:
        carrier = {}
        propagator.inject(carrier)

        # SQS message attributes for trace context
        message_attributes = {
            'traceparent': {
                'StringValue': carrier.get('traceparent', ''),
                'DataType': 'String'
            }
        }

        if 'tracestate' in carrier:
            message_attributes['tracestate'] = {
                'StringValue': carrier['tracestate'],
                'DataType': 'String'
            }

        sqs.send_message(
            QueueUrl=queue_url,
            MessageBody=message_body,
            MessageAttributes=message_attributes
        )

# Consuming from SQS
def process_sqs_message(message):
    # Extract trace context from SQS message attributes
    carrier = {}
    if 'MessageAttributes' in message:
        attrs = message['MessageAttributes']
        if 'traceparent' in attrs:
            carrier['traceparent'] = attrs['traceparent']['StringValue']
        if 'tracestate' in attrs:
            carrier['tracestate'] = attrs['tracestate']['StringValue']

    ctx = propagator.extract(carrier)
    span_context = trace.get_current_span(ctx).get_span_context()

    links = [Link(span_context)] if span_context.is_valid else []

    with tracer.start_as_current_span(
        "sqs_process",
        links=links
    ) as span:
        # Process message
        body = message['Body']
        return handle_message(body)

import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.SpanBuilder;
import io.opentelemetry.api.trace.SpanContext;
import io.opentelemetry.api.trace.Tracer;
import io.opentelemetry.context.Context;
import io.opentelemetry.context.propagation.TextMapGetter;
import io.opentelemetry.context.propagation.TextMapPropagator;
import io.opentelemetry.context.propagation.TextMapSetter;

public class MessageProcessor {
    private final Tracer tracer;
    private final TextMapPropagator propagator;

    // Producer: Publishing a message with trace context
    public void publishMessage(Queue queue, String messageBody) {
        Span span = tracer.spanBuilder("publish_message")
            .startSpan();

        try (var scope = span.makeCurrent()) {
            // Inject trace context into message headers
            Map<String, String> headers = new HashMap<>();
            propagator.inject(Context.current(), headers,
                (carrier, key, value) -> carrier.put(key, value));

            Message message = new Message(messageBody, headers);
            queue.publish(message);

            span.setAttribute("messaging.destination", queue.getName());
            span.setAttribute("messaging.system", "custom_queue");
        } finally {
            span.end();
        }
    }

    // Consumer: Processing a message with span link
    public void processMessage(Message message) {
        // Extract trace context from message headers
        Context extractedContext = propagator.extract(
            Context.current(),
            message.getHeaders(),
            (carrier, key) -> carrier.get(key)
        );

        // Get the span context from extracted context
        SpanContext producerSpanContext = Span.fromContext(extractedContext)
            .getSpanContext();

        // Create span with link to producer
        SpanBuilder spanBuilder = tracer.spanBuilder("process_message");

        if (producerSpanContext.isValid()) {
            spanBuilder.addLink(producerSpanContext);
        }

        Span span = spanBuilder.startSpan();

        try (var scope = span.makeCurrent()) {
            handleMessage(message.getBody());

            span.setAttribute("messaging.system", "custom_queue");
            span.setAttribute("messaging.operation", "process");
        } finally {
            span.end();
        }
    }

    // AWS SQS example
    public void processSQSMessage(
        software.amazon.awssdk.services.sqs.model.Message sqsMessage
    ) {
        Map<String, String> carrier = new HashMap<>();

        // Extract trace context from SQS message attributes
        sqsMessage.messageAttributes().forEach((key, value) -> {
            if (key.equals("traceparent") || key.equals("tracestate")) {
                carrier.put(key, value.stringValue());
            }
        });

        Context extractedContext = propagator.extract(
            Context.current(),
            carrier,
            (c, k) -> c.get(k)
        );

        SpanContext producerSpanContext = Span.fromContext(extractedContext)
            .getSpanContext();

        SpanBuilder spanBuilder = tracer.spanBuilder("sqs_process");

        if (producerSpanContext.isValid()) {
            spanBuilder.addLink(producerSpanContext);
        }

        Span span = spanBuilder.startSpan();

        try (var scope = span.makeCurrent()) {
            handleMessage(sqsMessage.body());
            span.setAttribute("messaging.system", "AmazonSQS");
        } finally {
            span.end();
        }
    }
}

const { trace, context, SpanKind } = require('@opentelemetry/api');
const { W3CTraceContextPropagator } = require('@opentelemetry/core');

const tracer = trace.getTracer('message-processor');
const propagator = new W3CTraceContextPropagator();

// Producer: Publishing a message with trace context
async function publishMessage(queue, messageBody) {
  const span = tracer.startSpan('publish_message', {
    kind: SpanKind.PRODUCER
  });

  return context.with(trace.setSpan(context.active(), span), async () => {
    try {
      // Inject trace context into message headers
      const carrier = {};
      propagator.inject(
        context.active(),
        carrier,
        {
          set: (carrier, key, value) => {
            carrier[key] = value;
          }
        }
      );

      const message = {
        body: messageBody,
        headers: carrier
      };

      await queue.publish(message);

      span.setAttribute('messaging.destination', queue.name);
      span.setAttribute('messaging.system', 'custom_queue');
    } finally {
      span.end();
    }
  });
}

// Consumer: Processing a message with span link
async function processMessage(message) {
  // Extract trace context from message headers
  const extractedContext = propagator.extract(
    context.active(),
    message.headers || {},
    {
      get: (carrier, key) => carrier[key]
    }
  );

  // Get the span context from extracted context
  const producerSpan = trace.getSpan(extractedContext);
  const producerSpanContext = producerSpan?.spanContext();

  // Create span with link
  const links = [];
  if (producerSpanContext && trace.isSpanContextValid(producerSpanContext)) {
    links.push({
      context: producerSpanContext
    });
  }

  const span = tracer.startSpan(
    'process_message',
    {
      kind: SpanKind.CONSUMER,
      links: links
    }
  );

  return context.with(trace.setSpan(context.active(), span), async () => {
    try {
      await handleMessage(message.body);

      span.setAttribute('messaging.system', 'custom_queue');
      span.setAttribute('messaging.operation', 'process');
    } finally {
      span.end();
    }
  });
}

// AWS SQS example using AWS SDK v3
const { SQSClient, SendMessageCommand, ReceiveMessageCommand } = require('@aws-sdk/client-sqs');

async function sendSQSMessage(queueUrl, messageBody) {
  const span = tracer.startSpan('sqs_publish');

  return context.with(trace.setSpan(context.active(), span), async () => {
    try {
      const carrier = {};
      propagator.inject(context.active(), carrier, {
        set: (c, k, v) => { c[k] = v; }
      });

      const messageAttributes = {
        traceparent: {
          StringValue: carrier.traceparent || '',
          DataType: 'String'
        }
      };

      if (carrier.tracestate) {
        messageAttributes.tracestate = {
          StringValue: carrier.tracestate,
          DataType: 'String'
        };
      }

      const client = new SQSClient({});
      await client.send(new SendMessageCommand({
        QueueUrl: queueUrl,
        MessageBody: messageBody,
        MessageAttributes: messageAttributes
      }));
    } finally {
      span.end();
    }
  });
}

async function processSQSMessage(message) {
  const carrier = {};

  if (message.MessageAttributes) {
    if (message.MessageAttributes.traceparent) {
      carrier.traceparent = message.MessageAttributes.traceparent.StringValue;
    }
    if (message.MessageAttributes.tracestate) {
      carrier.tracestate = message.MessageAttributes.tracestate.StringValue;
    }
  }

  const extractedContext = propagator.extract(context.active(), carrier, {
    get: (c, k) => c[k]
  });

  const producerSpanContext = trace.getSpan(extractedContext)?.spanContext();
  const links = producerSpanContext && trace.isSpanContextValid(producerSpanContext)
    ? [{ context: producerSpanContext }]
    : [];

  const span = tracer.startSpan('sqs_process', { links });

  return context.with(trace.setSpan(context.active(), span), async () => {
    try {
      await handleMessage(message.Body);
      span.setAttribute('messaging.system', 'AmazonSQS');
    } finally {
      span.end();
    }
  });
}

package main

import (
    "context"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/propagation"
    "go.opentelemetry.io/otel/trace"
)

var (
    tracer     = otel.Tracer("message-processor")
    propagator = propagation.TraceContext{}
)

// Producer: Publishing a message with trace context
func publishMessage(ctx context.Context, queue Queue, messageBody string) error {
    ctx, span := tracer.Start(ctx, "publish_message")
    defer span.End()

    // Inject trace context into message headers
    carrier := propagation.MapCarrier{}
    propagator.Inject(ctx, carrier)

    message := Message{
        Body:    messageBody,
        Headers: map[string]string(carrier),
    }

    err := queue.Publish(message)

    span.SetAttributes(
        attribute.String("messaging.destination", queue.Name()),
        attribute.String("messaging.system", "custom_queue"),
    )

    return err
}

// Consumer: Processing a message with span link
func processMessage(ctx context.Context, message Message) error {
    // Extract trace context from message headers
    carrier := propagation.MapCarrier(message.Headers)
    extractedCtx := propagator.Extract(ctx, carrier)

    // Get the span context from extracted context
    producerSpanContext := trace.SpanContextFromContext(extractedCtx)

    // Create span with link
    var links []trace.Link
    if producerSpanContext.IsValid() {
        links = []trace.Link{
            {
                SpanContext: producerSpanContext,
            },
        }
    }

    ctx, span := tracer.Start(
        ctx,
        "process_message",
        trace.WithLinks(links...),
    )
    defer span.End()

    err := handleMessage(message.Body)

    span.SetAttributes(
        attribute.String("messaging.system", "custom_queue"),
        attribute.String("messaging.operation", "process"),
    )

    return err
}

// AWS SQS example
func processSQSMessage(ctx context.Context, sqsMessage *sqs.Message) error {
    // Extract trace context from SQS message attributes
    carrier := propagation.MapCarrier{}

    if sqsMessage.MessageAttributes != nil {
        if tp, ok := sqsMessage.MessageAttributes["traceparent"]; ok {
            carrier["traceparent"] = *tp.StringValue
        }
        if ts, ok := sqsMessage.MessageAttributes["tracestate"]; ok {
            carrier["tracestate"] = *ts.StringValue
        }
    }

    extractedCtx := propagator.Extract(ctx, carrier)
    producerSpanContext := trace.SpanContextFromContext(extractedCtx)

    var links []trace.Link
    if producerSpanContext.IsValid() {
        links = []trace.Link{
            {SpanContext: producerSpanContext},
        }
    }

    ctx, span := tracer.Start(
        ctx,
        "sqs_process",
        trace.WithLinks(links...),
    )
    defer span.End()

    err := handleMessage(*sqsMessage.Body)

    span.SetAttributes(
        attribute.String("messaging.system", "AmazonSQS"),
    )

    return err
}

using System.Diagnostics;
using OpenTelemetry;
using OpenTelemetry.Context.Propagation;

public class MessageProcessor
{
    private static readonly ActivitySource ActivitySource = new("MessageProcessor");
    private static readonly TextMapPropagator Propagator = Propagators.DefaultTextMapPropagator;

    // Producer: Publishing a message with trace context
    public async Task PublishMessage(IQueue queue, string messageBody)
    {
        using var activity = ActivitySource.StartActivity("publish_message", ActivityKind.Producer);

        // Inject trace context into message headers
        var carrier = new Dictionary<string, string>();
        Propagator.Inject(
            new PropagationContext(activity.Context, Baggage.Current),
            carrier,
            (c, key, value) => c[key] = value
        );

        var message = new Message
        {
            Body = messageBody,
            Headers = carrier
        };

        await queue.PublishAsync(message);

        activity?.SetTag("messaging.destination", queue.Name);
        activity?.SetTag("messaging.system", "custom_queue");
    }

    // Consumer: Processing a message with span link
    public async Task ProcessMessage(Message message)
    {
        // Extract trace context from message headers
        var propagationContext = Propagator.Extract(
            default,
            message.Headers ?? new Dictionary<string, string>(),
            (carrier, key) => carrier.TryGetValue(key, out var value) ? new[] { value } : Array.Empty<string>()
        );

        var producerContext = propagationContext.ActivityContext;

        // Create span with link
        var links = new List<ActivityLink>();
        if (producerContext != default)
        {
            links.Add(new ActivityLink(producerContext));
        }

        using var activity = ActivitySource.StartActivity(
            "process_message",
            ActivityKind.Consumer,
            parentContext: default,
            links: links
        );

        await HandleMessage(message.Body);

        activity?.SetTag("messaging.system", "custom_queue");
        activity?.SetTag("messaging.operation", "process");
    }

    // AWS SQS example
    public async Task ProcessSQSMessage(Amazon.SQS.Model.Message sqsMessage)
    {
        // Extract trace context from SQS message attributes
        var carrier = new Dictionary<string, string>();

        if (sqsMessage.MessageAttributes != null)
        {
            if (sqsMessage.MessageAttributes.TryGetValue("traceparent", out var tp))
            {
                carrier["traceparent"] = tp.StringValue;
            }
            if (sqsMessage.MessageAttributes.TryGetValue("tracestate", out var ts))
            {
                carrier["tracestate"] = ts.StringValue;
            }
        }

        var propagationContext = Propagator.Extract(
            default,
            carrier,
            (c, key) => c.TryGetValue(key, out var value) ? new[] { value } : Array.Empty<string>()
        );

        var links = new List<ActivityLink>();
        if (propagationContext.ActivityContext != default)
        {
            links.Add(new ActivityLink(propagationContext.ActivityContext));
        }

        using var activity = ActivitySource.StartActivity(
            "sqs_process",
            ActivityKind.Consumer,
            parentContext: default,
            links: links
        );

        await HandleMessage(sqsMessage.Body);

        activity?.SetTag("messaging.system", "AmazonSQS");
    }
}

スパンリンクのベストプラクティス

スパンリンクを実装する場合は、次のベストプラクティスに従ってください。

Always propagate trace context: W3Ct レースコンテキスト (traceparentおよびtracestateヘッダー) がメッセージヘッダーまたはメタデータに含まれていることを確認してください。
Validate span context: スパンリンクを作成する前に、抽出されたスパンコンテキストが有効かどうかを常に確認します。無効なコンテキストではリンクを作成しないでください。
Use appropriate span kinds: メッセージ公開スパンの種類をPRODUCERに設定し、メッセージ処理スパンの種類をCONSUMER設定します。
Add messaging attributes: コンテキストを提供するために、メッセージングシステムのセマンティック規則 (messaging.system 、 messaging.destination 、 messaging.operationなど) を含めます。
Consider sampling: リンクされたトレースは両方ともサンプリングしないと New Relic に表示されません。スパンリンクを使用する重要なワークフローのサンプリング戦略を調整します。
Handle batch processing carefully: バッチ処理されたメッセージを処理する場合、追跡可能性を維持するために、メッセージごとに個別のスパンリンクを作成します。

New Relicでスパンリンクを表示する

ダッシュボードにスパンリンクを実装したら、 New Relic UIでそれらを表示および操作できます。

トレースのトレース詳細ページに移動します。
フィルターバーのスパンリンクバッジを探します。これはリンクのあるスパンの数を示します。
リンクのあるスパンを選択すると、スパンの詳細ペインのSpan linksタブが表示されます。
リンクされたトレースをクリックすると、関連するトレース間を移動できます。

UI でのスパンリンクの使用の詳細については、「スパンリンクについて」を参照してください。

この機械翻訳は、参考として提供されています。

New Relic の OpenTelemetry トレース

OTLPスパンマッピング

表の脚注

スパンのリンク

スパンリンクを使用する場合

スパンリンクの実装

Python

Java

Node.js

Go

.NET

スパンリンクのベストプラクティス

New Relicでスパンリンクを表示する

この機械翻訳は、参考として提供されています。

New Relic の OpenTelemetry トレース

OTLPスパンマッピング .css-21sua1{background:none;border:none;width:0;padding:0;}

表の脚注

スパンのリンク

スパンリンクを使用する場合

スパンリンクの実装

Java

Node.js

Go

.NET

スパンリンクのベストプラクティス

New Relicでスパンリンクを表示する

OTLPスパンマッピング