/* * Copyright The OpenTelemetry Authors * SPDX-License-Identifier: Apache-2.0 */ import io.netty.bootstrap.Bootstrap import io.netty.buffer.Unpooled import io.netty.channel.ChannelInitializer import io.netty.channel.ChannelPipeline import io.netty.channel.EventLoopGroup import io.netty.channel.nio.NioEventLoopGroup import io.netty.channel.socket.SocketChannel import io.netty.channel.socket.nio.NioSocketChannel import io.netty.handler.codec.http.DefaultFullHttpRequest import io.netty.handler.codec.http.HttpClientCodec import io.netty.handler.codec.http.HttpHeaderNames import io.netty.handler.codec.http.HttpMethod import io.netty.handler.codec.http.HttpVersion import io.opentelemetry.instrumentation.api.instrumenter.network.internal.NetworkAttributes import io.opentelemetry.instrumentation.api.internal.SemconvStability import io.opentelemetry.instrumentation.netty.v4_1.ClientHandler import io.opentelemetry.instrumentation.test.AgentTestTrait import io.opentelemetry.instrumentation.test.InstrumentationSpecification import io.opentelemetry.instrumentation.test.utils.PortUtils import io.opentelemetry.instrumentation.testing.junit.http.HttpClientTestServer import io.opentelemetry.semconv.trace.attributes.SemanticAttributes import spock.lang.Shared import java.util.concurrent.CompletableFuture import java.util.concurrent.TimeUnit import static io.opentelemetry.api.trace.SpanKind.CLIENT import static io.opentelemetry.api.trace.SpanKind.INTERNAL import static io.opentelemetry.api.trace.SpanKind.SERVER import static io.opentelemetry.api.trace.StatusCode.ERROR import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.NetTransportValues.IP_TCP class Netty41ConnectionSpanTest extends InstrumentationSpecification implements AgentTestTrait { @Shared private HttpClientTestServer server @Shared private EventLoopGroup eventLoopGroup = new NioEventLoopGroup() @Shared private Bootstrap bootstrap = buildBootstrap() def setupSpec() { server = new HttpClientTestServer(openTelemetry) server.start() } def cleanupSpec() { eventLoopGroup.shutdownGracefully() server.stop() } Bootstrap buildBootstrap() { Bootstrap bootstrap = new Bootstrap() bootstrap.group(eventLoopGroup) .channel(NioSocketChannel) .handler(new ChannelInitializer() { @Override protected void initChannel(SocketChannel socketChannel) throws Exception { ChannelPipeline pipeline = socketChannel.pipeline() pipeline.addLast(new HttpClientCodec()) } }) return bootstrap } DefaultFullHttpRequest buildRequest(String method, URI uri, Map headers) { def request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.valueOf(method), uri.path, Unpooled.EMPTY_BUFFER) request.headers().set(HttpHeaderNames.HOST, uri.host + ":" + uri.port) headers.each { k, v -> request.headers().set(k, v) } return request } int sendRequest(DefaultFullHttpRequest request, URI uri) { def channel = bootstrap.connect(uri.host, uri.port).sync().channel() def result = new CompletableFuture() channel.pipeline().addLast(new ClientHandler(result)) channel.writeAndFlush(request).get() return result.get(20, TimeUnit.SECONDS) } def "test successful request"() { when: def uri = URI.create("http://localhost:${server.httpPort()}/success") def request = buildRequest("GET", uri, [:]) def responseCode = runWithSpan("parent") { sendRequest(request, uri) } then: responseCode == 200 assertTraces(1) { trace(0, 5) { def list = Arrays.asList("RESOLVE", "CONNECT") spans.subList(1, 3).sort(Comparator.comparing { item -> list.indexOf(item.name) }) span(0) { name "parent" kind INTERNAL hasNoParent() } span(1) { name "RESOLVE" kind INTERNAL childOf span(0) if (SemconvStability.emitOldHttpSemconv()) { attributes { "$SemanticAttributes.NET_TRANSPORT" IP_TCP "$SemanticAttributes.NET_PEER_NAME" uri.host "$SemanticAttributes.NET_PEER_PORT" uri.port } } if (SemconvStability.emitStableHttpSemconv()) { attributes { "$NetworkAttributes.SERVER_ADDRESS" uri.host "$NetworkAttributes.SERVER_PORT" uri.port } } } span(2) { name "CONNECT" kind INTERNAL childOf(span(0)) if (SemconvStability.emitOldHttpSemconv()) { attributes { "$SemanticAttributes.NET_TRANSPORT" IP_TCP "$SemanticAttributes.NET_PEER_NAME" uri.host "$SemanticAttributes.NET_PEER_PORT" uri.port "$SemanticAttributes.NET_SOCK_PEER_ADDR" "127.0.0.1" } } if (SemconvStability.emitStableHttpSemconv()) { attributes { "$NetworkAttributes.NETWORK_TRANSPORT" "tcp" "$NetworkAttributes.NETWORK_TYPE" "ipv4" "$NetworkAttributes.SERVER_ADDRESS" uri.host "$NetworkAttributes.SERVER_PORT" uri.port "$NetworkAttributes.SERVER_SOCKET_ADDRESS" "127.0.0.1" } } } span(3) { name "GET" kind CLIENT childOf(span(0)) } span(4) { name "test-http-server" kind SERVER childOf(span(3)) } } } } def "test failing request"() { when: URI uri = URI.create("http://localhost:${PortUtils.UNUSABLE_PORT}") def request = buildRequest("GET", uri, [:]) runWithSpan("parent") { sendRequest(request, uri) } then: def thrownException = thrown(Exception) and: assertTraces(1) { trace(0, 3) { def list = Arrays.asList("RESOLVE", "CONNECT") spans.subList(1, 3).sort(Comparator.comparing { item -> list.indexOf(item.name) }) span(0) { name "parent" kind INTERNAL hasNoParent() status ERROR errorEvent(thrownException.class, thrownException.message) } span(1) { name "RESOLVE" kind INTERNAL childOf span(0) if (SemconvStability.emitOldHttpSemconv()) { attributes { "$SemanticAttributes.NET_TRANSPORT" IP_TCP "$SemanticAttributes.NET_PEER_NAME" uri.host "$SemanticAttributes.NET_PEER_PORT" uri.port } } if (SemconvStability.emitStableHttpSemconv()) { attributes { "$NetworkAttributes.SERVER_ADDRESS" uri.host "$NetworkAttributes.SERVER_PORT" uri.port } } } span(2) { name "CONNECT" kind INTERNAL childOf span(0) status ERROR errorEvent(thrownException.class, thrownException.message) if (SemconvStability.emitOldHttpSemconv()) { attributes { "$SemanticAttributes.NET_TRANSPORT" IP_TCP "$SemanticAttributes.NET_PEER_NAME" uri.host "$SemanticAttributes.NET_PEER_PORT" uri.port "$SemanticAttributes.NET_SOCK_PEER_ADDR" { it == "127.0.0.1" || it == null } } } if (SemconvStability.emitStableHttpSemconv()) { attributes { "$NetworkAttributes.NETWORK_TRANSPORT" "tcp" "$NetworkAttributes.NETWORK_TYPE" { it == "ipv4" || it == null } "$NetworkAttributes.SERVER_ADDRESS" uri.host "$NetworkAttributes.SERVER_PORT" uri.port "$NetworkAttributes.SERVER_SOCKET_ADDRESS" { it == "127.0.0.1" || it == null } } } } } } } }