diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/InternalSpanKind.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/InternalSpanKind.java index de34d8fa25d..70d70f1a33c 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/InternalSpanKind.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/InternalSpanKind.java @@ -23,11 +23,12 @@ import java.lang.annotation.Target; /** - * Annotation that marks a proxied client method as SpanKind.INTERNAL. + * Annotation that marks a method of a proxied delegate class, or every method declared in that class, as + * SpanKind.INTERNAL. * Used for methods that do not send an outbound RPC. */ @Retention(RetentionPolicy.RUNTIME) -@Target(ElementType.METHOD) +@Target({ElementType.METHOD, ElementType.TYPE}) public @interface InternalSpanKind { } diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/SpanAttributeKeys.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/SpanAttributeKeys.java new file mode 100644 index 00000000000..509e3ff5747 --- /dev/null +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/SpanAttributeKeys.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.hdds.tracing; + +/** + * Span attribute keys shared by Ozone services and clients. + * Fields that are also in the audit log (volume, bucket, key, ...) use the audit key unchanged instead. + */ +public final class SpanAttributeKeys { + + /** Status or result code of a failed request, for example KEY_NOT_FOUND. */ + public static final String STATUS = "ozone.status"; + + private SpanAttributeKeys() { + } +} diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/TraceAllMethod.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/TraceAllMethod.java index 2acb46d11f8..cf078cdbcbe 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/TraceAllMethod.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/TraceAllMethod.java @@ -19,8 +19,11 @@ import static java.util.Collections.emptyMap; +import io.opentelemetry.api.trace.Span; import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.context.Scope; import java.lang.reflect.InvocationHandler; +import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.util.Arrays; import java.util.HashMap; @@ -40,10 +43,9 @@ public class TraceAllMethod implements InvocationHandler { */ private final Map[], DelegatedMethodInfo>> methods = new HashMap<>(); private final T delegate; - private final String name; - public TraceAllMethod(T delegate, String name) { + public TraceAllMethod(T delegate, String name, SpanKind spanKind) { this.delegate = delegate; this.name = name; for (Method method : delegate.getClass().getMethods()) { @@ -52,7 +54,7 @@ public TraceAllMethod(T delegate, String name) { } boolean shouldSkip = method.isAnnotationPresent(SkipTracing.class); methods.computeIfAbsent(method.getName(), any -> new HashMap<>()) - .put(method.getParameterTypes(), new DelegatedMethodInfo(shouldSkip, method)); + .put(method.getParameterTypes(), new DelegatedMethodInfo(shouldSkip, findSpanKind(method, spanKind))); } } @@ -60,40 +62,35 @@ public TraceAllMethod(T delegate, String name) { public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { DelegatedMethodInfo methodInfo = findDelegatedMethod(method); - if (methodInfo == null) { - throw new NoSuchMethodException("Method not found: " + method.getName()); + if (methodInfo == null || methodInfo.shouldSkip) { + return invokeDelegate(method, args); } - boolean shouldSkip = methodInfo.shouldSkip; - Method delegateMethod = methodInfo.delegateMethod; - if (shouldSkip) { - try { - return delegateMethod.invoke(delegate, args); - } catch (Exception ex) { - if (ex.getCause() != null) { - throw ex.getCause(); - } else { - throw ex; - } - } + + Span span = TracingUtil.createSpan(name + "." + method.getName(), methodInfo.spanKind); + try (Scope ignored = span.makeCurrent()) { + return invokeDelegate(method, args); + } catch (Throwable ex) { + TracingUtil.recordException(span, ex); + throw ex; + } finally { + span.end(); } + } - // if a call is within a process, not an outbound RPC to other processes (OM/SCM/DN) , it is marked as INTERNAL. - SpanKind spanKind = delegateMethod.isAnnotationPresent(InternalSpanKind.class) - ? SpanKind.INTERNAL - : SpanKind.CLIENT; + private Object invokeDelegate(Method method, Object[] args) throws Throwable { + try { + return method.invoke(delegate, args); + } catch (InvocationTargetException e) { + throw e.getCause(); + } + } - try (TracingUtil.TraceCloseable ignored = TracingUtil.createActivatedSpan( - name + "." + method.getName(), spanKind)) { - try { - return delegateMethod.invoke(delegate, args); - } catch (Exception ex) { - if (ex.getCause() != null) { - throw ex.getCause(); - } else { - throw ex; - } - } + private static SpanKind findSpanKind(Method method, SpanKind defaultSpanKind) { + if (method.isAnnotationPresent(InternalSpanKind.class) + || method.getDeclaringClass().isAnnotationPresent(InternalSpanKind.class)) { + return SpanKind.INTERNAL; } + return defaultSpanKind; } private DelegatedMethodInfo findDelegatedMethod(Method method) { @@ -111,11 +108,11 @@ private DelegatedMethodInfo findDelegatedMethod(Method method) { */ private static final class DelegatedMethodInfo { private final boolean shouldSkip; - private final Method delegateMethod; + private final SpanKind spanKind; - private DelegatedMethodInfo(boolean shouldSkip, Method delegateMethod) { + private DelegatedMethodInfo(boolean shouldSkip, SpanKind spanKind) { this.shouldSkip = shouldSkip; - this.delegateMethod = delegateMethod; + this.spanKind = spanKind; } } } diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/TracingUtil.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/TracingUtil.java index b845c828996..3388167cc3e 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/TracingUtil.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/TracingUtil.java @@ -394,13 +394,34 @@ private static boolean hasUsableTracer() { */ public static T createProxy( T delegate, Class itf, ConfigurationSource conf) { + return createProxy(delegate, itf, conf, SpanKind.INTERNAL); + } + + /** Same as {@link #createProxy(Object, Class, ConfigurationSource)}, with CLIENT spans for an RPC delegate. */ + public static T createClientProxy(T delegate, Class itf, ConfigurationSource conf) { + return createProxy(delegate, itf, conf, SpanKind.CLIENT); + } + + /** Creates a proxy with spans named {@code .}, also usable at startup without a span. */ + public static T createProxy(T delegate, Class itf, SpanKind spanKind) { + if (!tracingEnabled && !applicationAware) { + return delegate; + } + return newProxy(delegate, itf, delegate.getClass().getSimpleName(), spanKind); + } + + private static T createProxy(T delegate, Class itf, ConfigurationSource conf, SpanKind spanKind) { if (!shouldCreateTracingProxy(conf)) { return delegate; } + return newProxy(delegate, itf, itf.getSimpleName(), spanKind); + } + + private static T newProxy(T delegate, Class itf, String name, SpanKind spanKind) { Class aClass = delegate.getClass(); return itf.cast(Proxy.newProxyInstance(aClass.getClassLoader(), new Class[] {itf}, - new TraceAllMethod<>(delegate, itf.getSimpleName()))); + new TraceAllMethod<>(delegate, name, spanKind))); } public static boolean isTracingEnabled(ConfigurationSource conf) { @@ -499,7 +520,7 @@ private static R executeInSpan(Span span, return supplier.get(); } catch (Exception ex) { span.addEvent("Failed with exception: " + ex.getMessage()); - span.setStatus(StatusCode.ERROR); + recordException(span, ex); throw ex; } finally { span.end(); @@ -515,7 +536,7 @@ private static void executeInSpan(Span span, runnable.run(); } catch (Exception ex) { span.addEvent("Failed with exception: " + ex.getMessage()); - span.setStatus(StatusCode.ERROR); + recordException(span, ex); throw ex; } finally { span.end(); @@ -556,6 +577,63 @@ public static Span getActiveSpan() { return Span.current(); } + public static Span createSpan(String spanName) { + return createSpan(spanName, SpanKind.INTERNAL); + } + + public static Span createSpan(String spanName, SpanKind spanKind) { + return buildSpan(spanName, spanKind); + } + + public static void recordExceptionInCurrentSpan(Throwable t) { + recordException(Span.current(), t); + } + + public static void recordException(Span span, Throwable t) { + if (t == null || !span.isRecording()) { + return; + } + span.recordException(t); + if (t.getMessage() == null) { + span.setStatus(StatusCode.ERROR); + } else { + span.setStatus(StatusCode.ERROR, t.getMessage()); + } + } + + public static void addAttributesToCurrentSpan(String key, String value) { + Span span = Span.current(); + if (value != null && span.isRecording()) { + span.setAttribute(key, value); + } + } + + public static void addAttributesToCurrentSpan(String key, long value) { + Span span = Span.current(); + if (span.isRecording()) { + span.setAttribute(key, value); + } + } + + public static void addAttributesToCurrentSpan(String key, boolean value) { + Span span = Span.current(); + if (span.isRecording()) { + span.setAttribute(key, value); + } + } + + public static void addEventToCurrentSpan(String name, String key, String value) { + Span span = Span.current(); + if (!span.isRecording()) { + return; + } + if (value == null) { + span.addEvent(name); + } else { + span.addEvent(name, Attributes.of(AttributeKey.stringKey(key), value)); + } + } + /** * AutoCloseable interface for tracing span but no exception is thrown in close. */ diff --git a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/tracing/TestSpanCollector.java b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/tracing/TestSpanCollector.java new file mode 100644 index 00000000000..e8f91b4313b --- /dev/null +++ b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/tracing/TestSpanCollector.java @@ -0,0 +1,166 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.hdds.tracing; + +import static io.opentelemetry.api.common.AttributeKey.booleanKey; +import static io.opentelemetry.api.common.AttributeKey.longKey; +import static io.opentelemetry.api.common.AttributeKey.stringKey; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; + +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.StatusCode; +import io.opentelemetry.context.Context; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.trace.ReadWriteSpan; +import io.opentelemetry.sdk.trace.ReadableSpan; +import io.opentelemetry.sdk.trace.SdkTracerProvider; +import io.opentelemetry.sdk.trace.SpanProcessor; +import io.opentelemetry.sdk.trace.data.SpanData; +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.stream.Collectors; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; + +/** + * {@link SpanProcessor} that collects ended spans in memory, so tests can assert on them. + */ +public final class TestSpanCollector implements SpanProcessor, AutoCloseable { + private final List spans = new CopyOnWriteArrayList<>(); + + private TestSpanCollector() { + } + + /** Registers a collecting SDK as the global OpenTelemetry and initializes Ozone tracing with it. */ + public static TestSpanCollector install(String serviceName) { + TracingUtil.shutdownTracing(); + GlobalOpenTelemetry.resetForTest(); + TestSpanCollector collector = new TestSpanCollector(); + GlobalOpenTelemetry.set(OpenTelemetrySdk.builder() + .setTracerProvider(SdkTracerProvider.builder().addSpanProcessor(collector).build()) + .build()); + OzoneConfiguration conf = new OzoneConfiguration(); + conf.setBoolean("ozone.tracing.enabled", true); + TracingUtil.initTracing(serviceName, conf); + return collector; + } + + public List getSpans() { + return spans; + } + + public List named(String name) { + return spans.stream().filter(span -> span.getName().equals(name)).collect(Collectors.toList()); + } + + /** Fails unless exactly one ended span is named {@code spanName}. */ + public void assertSingleSpan(String spanName) { + onlySpan(spanName); + } + + public void assertError(String spanName) { + assertEquals(StatusCode.ERROR, onlySpan(spanName).getStatus().getStatusCode(), () -> spanName + " status"); + } + + public void assertError(String spanName, String expectedDescription) { + SpanData span = onlySpan(spanName); + assertEquals(StatusCode.ERROR, span.getStatus().getStatusCode(), () -> spanName + " status"); + assertEquals(expectedDescription, span.getStatus().getDescription(), () -> spanName + " status description"); + } + + public void assertKind(String spanName, SpanKind expected) { + assertEquals(expected, onlySpan(spanName).getKind(), () -> spanName + " kind"); + } + + public void assertNotError(String spanName) { + assertNotEquals(StatusCode.ERROR, onlySpan(spanName).getStatus().getStatusCode(), () -> spanName + " status"); + } + + public void assertAttribute(String spanName, String key, String expected) { + assertEquals(expected, onlySpan(spanName).getAttributes().get(stringKey(key)), () -> spanName + " " + key); + } + + public void assertAttribute(String spanName, String key, long expected) { + assertEquals(expected, onlySpan(spanName).getAttributes().get(longKey(key)), () -> spanName + " " + key); + } + + public void assertAttribute(String spanName, String key, boolean expected) { + assertEquals(expected, onlySpan(spanName).getAttributes().get(booleanKey(key)), () -> spanName + " " + key); + } + + /** Fails if the span has {@code key} with a value of any type. */ + public void assertNoAttribute(String spanName, String key) { + Set keys = onlySpan(spanName).getAttributes().asMap().keySet().stream() + .map(AttributeKey::getKey) + .collect(Collectors.toSet()); + assertFalse(keys.contains(key), () -> spanName + " has attribute " + key + ": " + keys); + } + + /** Fails unless the span recorded exactly one exception event, of type {@code type}. */ + public void assertExceptionRecorded(String spanName, Class type) { + List types = onlySpan(spanName).getEvents().stream() + .filter(event -> "exception".equals(event.getName())) + .map(event -> event.getAttributes().get(stringKey("exception.type"))) + .collect(Collectors.toList()); + assertEquals(Collections.singletonList(type.getName()), types, () -> spanName + " exception events"); + } + + /** Fails unless the span named {@code childName} is a direct child of the span named {@code parentName}. */ + public void assertParent(String childName, String parentName) { + SpanData child = onlySpan(childName); + SpanData parent = onlySpan(parentName); + assertEquals(parent.getTraceId(), child.getTraceId(), () -> childName + " trace ID"); + assertEquals(parent.getSpanId(), child.getParentSpanId(), () -> childName + " parent"); + } + + private SpanData onlySpan(String spanName) { + List matches = named(spanName); + assertEquals(1, matches.size(), () -> "spans named " + spanName + ": " + spans); + return matches.get(0).toSpanData(); + } + + @Override + public void close() { + TracingUtil.shutdownTracing(); + GlobalOpenTelemetry.resetForTest(); + } + + @Override + public void onStart(Context parentContext, ReadWriteSpan span) { + } + + @Override + public boolean isStartRequired() { + return false; + } + + @Override + public void onEnd(ReadableSpan span) { + spans.add(span); + } + + @Override + public boolean isEndRequired() { + return true; + } +} diff --git a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/tracing/TestTraceAllMethod.java b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/tracing/TestTraceAllMethod.java index 8c10861d931..6958a04d85c 100644 --- a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/tracing/TestTraceAllMethod.java +++ b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/tracing/TestTraceAllMethod.java @@ -18,12 +18,19 @@ package org.apache.hadoop.hdds.tracing; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import io.opentelemetry.api.trace.Span; import io.opentelemetry.api.trace.SpanKind; import io.opentelemetry.sdk.trace.ReadableSpan; import java.io.IOException; +import java.lang.reflect.Modifier; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.stream.Collectors; import org.junit.jupiter.api.Test; /** @@ -33,21 +40,100 @@ public class TestTraceAllMethod { @Test public void testUnknownMethod() { - TraceAllMethod subject = new TraceAllMethod<>(new ServiceImpl(), - "service"); + TraceAllMethod subject = new TraceAllMethod<>(new ServiceImpl(), "service", SpanKind.CLIENT); assertThrows(NoSuchMethodException.class, () -> subject.invoke(null, ServiceImpl.class.getMethod("no such method"), new Object[]{})); } @Test public void testDefaultMethod() throws Throwable { - TraceAllMethod subject = new TraceAllMethod<>(new ServiceImpl(), - "service"); + TraceAllMethod subject = new TraceAllMethod<>(new ServiceImpl(), "service", SpanKind.CLIENT); assertEquals("Hello default", subject.invoke(null, ServiceImpl.class.getMethod("defaultMethod"), new Object[]{})); } + @Test + public void testConstructorKindIsUsed() throws Throwable { + try (TestSpanCollector spans = TestSpanCollector.install("test")) { + TraceAllMethod subject = new TraceAllMethod<>(new ServiceImpl(), "service", SpanKind.CLIENT); + assertEquals("normal", subject.invoke(null, Service.class.getMethod("normalMethod"), null)); + spans.assertKind("service.normalMethod", SpanKind.CLIENT); + } + } + + @Test + public void testMethodAnnotationOverridesConstructorKind() throws Throwable { + try (TestSpanCollector spans = TestSpanCollector.install("test")) { + TraceAllMethod subject = new TraceAllMethod<>(new ServiceImpl(), "service", SpanKind.SERVER); + assertEquals("normal", subject.invoke(null, Service.class.getMethod("normalMethod"), null)); + assertEquals("local", subject.invoke(null, Service.class.getMethod("localMethod"), null)); + spans.assertKind("service.normalMethod", SpanKind.SERVER); + spans.assertKind("service.localMethod", SpanKind.INTERNAL); + } + } + + @Test + public void testClassAnnotationSkipsInheritedMethods() throws Throwable { + try (TestSpanCollector spans = TestSpanCollector.install("test")) { + TraceAllMethod subject = new TraceAllMethod<>(new InternalServiceImpl(), "service", SpanKind.CLIENT); + assertEquals("normal", subject.invoke(null, Service.class.getMethod("normalMethod"), null)); + assertEquals("Internal x", subject.invoke(null, Service.class.getMethod("otherMethod", String.class), + new Object[]{"x"})); + spans.assertKind("service.normalMethod", SpanKind.CLIENT); + spans.assertKind("service.otherMethod", SpanKind.INTERNAL); + } + } + + @Test + public void testOverloadsResolveTheirOwnKind() throws Throwable { + try (TestSpanCollector spans = TestSpanCollector.install("test")) { + TraceAllMethod subject = + new TraceAllMethod<>(new OverloadedServiceImpl(), "service", SpanKind.CLIENT); + assertEquals("a", subject.invoke(null, OverloadedService.class.getMethod("call", String.class), + new Object[]{"a"})); + assertEquals("a2", subject.invoke(null, OverloadedService.class.getMethod("call", String.class, int.class), + new Object[]{"a", 2})); + assertEquals(Arrays.asList(SpanKind.CLIENT, SpanKind.INTERNAL), + spans.named("service.call").stream().map(ReadableSpan::getKind).collect(Collectors.toList())); + } + } + + @Test + public void testExceptionIsRecorded() { + try (TestSpanCollector spans = TestSpanCollector.install("test")) { + TraceAllMethod subject = new TraceAllMethod<>(new ServiceImpl(), "service", + SpanKind.CLIENT); + IOException ex = assertThrows(IOException.class, + () -> subject.invoke(null, Service.class.getMethod("tracedThrowingMethod"), null)); + assertEquals("Failed", ex.getMessage()); + spans.assertError("service.tracedThrowingMethod", "Failed"); + spans.assertExceptionRecorded("service.tracedThrowingMethod", IOException.class); + } + } + + @Test + public void testNonPublicDelegateClass() throws Throwable { + try (TestSpanCollector spans = TestSpanCollector.install("test")) { + List delegate = Collections.unmodifiableList(Arrays.asList("a", "b")); + assertFalse(Modifier.isPublic(delegate.getClass().getModifiers())); + TraceAllMethod> subject = new TraceAllMethod<>(delegate, "List", SpanKind.INTERNAL); + assertEquals(2, subject.invoke(null, List.class.getMethod("size"), null)); + spans.assertKind("List.size", SpanKind.INTERNAL); + } + } + + @Test + public void testObjectMethodsGoToDelegate() throws Throwable { + try (TestSpanCollector spans = TestSpanCollector.install("test")) { + ServiceImpl delegate = new ServiceImpl(); + TraceAllMethod subject = new TraceAllMethod<>(delegate, "service", SpanKind.CLIENT); + assertEquals(delegate.toString(), subject.invoke(null, Object.class.getMethod("toString"), null)); + assertEquals(delegate.hashCode(), subject.invoke(null, Object.class.getMethod("hashCode"), null)); + assertTrue(spans.getSpans().isEmpty()); + } + } + /** * Example interface for testing {@link TraceAllMethod}. */ @@ -58,10 +144,15 @@ default String defaultMethod() { String localMethod(); + @InternalSpanKind + String annotatedOnlyOnInterface(); + String skippedMethod(); void throwingMethod() throws IOException; + void tracedThrowingMethod() throws IOException; + String normalMethod(); String otherMethod(String name); @@ -98,6 +189,11 @@ public void throwingMethod() throws IOException { throw new IOException("Original Exception"); } + @Override + public void tracedThrowingMethod() throws IOException { + throw new IOException("Failed"); + } + @Override public String normalMethod() { this.spanActive = Span.current().getSpanContext().isValid(); @@ -115,8 +211,49 @@ public String localMethod() { return "local"; } + @Override + public String annotatedOnlyOnInterface() { + return "interfaceLocal"; + } + public boolean wasSpanActive() { return spanActive; } } + + /** + * Annotated subclass: only the methods it declares itself are INTERNAL. + */ + @InternalSpanKind + public static class InternalServiceImpl extends ServiceImpl { + @Override + public String otherMethod(String name) { + return "Internal " + name; + } + } + + /** + * Interface with an overloaded method. + */ + public interface OverloadedService { + String call(String name); + + String call(String name, int count); + } + + /** + * Implementation where only one overload is INTERNAL. + */ + public static class OverloadedServiceImpl implements OverloadedService { + @Override + public String call(String name) { + return name; + } + + @Override + @InternalSpanKind + public String call(String name, int count) { + return name + count; + } + } } diff --git a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/tracing/TestTracingAnnotations.java b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/tracing/TestTracingAnnotations.java index 823758ff93e..30d93af44a0 100644 --- a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/tracing/TestTracingAnnotations.java +++ b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/tracing/TestTracingAnnotations.java @@ -17,18 +17,26 @@ package org.apache.hadoop.hdds.tracing; +import static org.apache.hadoop.hdds.tracing.TracingUtil.createClientProxy; import static org.apache.hadoop.hdds.tracing.TracingUtil.createProxy; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import io.opentelemetry.api.GlobalOpenTelemetry; import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.trace.SdkTracerProvider; +import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor; +import io.opentelemetry.sdk.trace.export.SpanExporter; import java.io.IOException; import org.apache.hadoop.hdds.conf.InMemoryConfigurationForTesting; import org.apache.hadoop.hdds.conf.MutableConfigurationSource; import org.apache.hadoop.hdds.tracing.TestTraceAllMethod.Service; import org.apache.hadoop.hdds.tracing.TestTraceAllMethod.ServiceImpl; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; /** @@ -37,6 +45,31 @@ */ public class TestTracingAnnotations { + @BeforeEach + public void resetGlobalState() { + TracingUtil.shutdownTracing(); + GlobalOpenTelemetry.resetForTest(); + } + + /** + * Tear down tracing state after each test. + */ + @AfterEach + public void cleanup() { + TracingUtil.shutdownTracing(); + GlobalOpenTelemetry.resetForTest(); + } + + /** + * Puts a real GlobalOpenTelemetry in place with no exporter, so tests stay offline. + */ + private static void installNoExportGlobalOpenTelemetry() { + SdkTracerProvider provider = + SdkTracerProvider.builder().addSpanProcessor(SimpleSpanProcessor.create(SpanExporter.noop())).build(); + OpenTelemetrySdk sdk = OpenTelemetrySdk.builder().setTracerProvider(provider).build(); + GlobalOpenTelemetry.set(sdk); + } + private static MutableConfigurationSource tracingEnabled() { MutableConfigurationSource config = new InMemoryConfigurationForTesting(); config.setBoolean("ozone.tracing.enabled", true); @@ -67,6 +100,7 @@ public void testSkipTracingExceptionUnwrapped() { @Test public void testProxyNormalVsSkipped() { + installNoExportGlobalOpenTelemetry(); TracingUtil.initTracing("TestService", tracingEnabled()); ServiceImpl impl = new ServiceImpl(); Service serviceProxy = createProxy(impl, Service.class, tracingEnabled()); @@ -77,12 +111,70 @@ public void testProxyNormalVsSkipped() { @Test public void testInternalSpanKind() { + installNoExportGlobalOpenTelemetry(); TracingUtil.initTracing("TestService", tracingEnabled()); ServiceImpl impl = new ServiceImpl(); - Service serviceProxy = createProxy(impl, Service.class, tracingEnabled()); + Service serviceProxy = createClientProxy(impl, Service.class, tracingEnabled()); serviceProxy.localMethod(); assertTrue(impl.wasSpanActive(), "Local method should have an active span."); assertEquals(SpanKind.INTERNAL, impl.lastSpanKind(), "@InternalSpanKind method should use INTERNAL."); } + @Test + public void testInterfaceMethodAnnotationIsIgnored() { + try (TestSpanCollector spans = TestSpanCollector.install("TestService")) { + Service serviceProxy = createClientProxy(new ServiceImpl(), Service.class, tracingEnabled()); + serviceProxy.annotatedOnlyOnInterface(); + spans.assertKind("Service.annotatedOnlyOnInterface", SpanKind.CLIENT); + } + } + + @Test + public void testLambdaDelegateUsesDefaultKind() { + try (TestSpanCollector spans = TestSpanCollector.install("TestService")) { + LocalService serviceProxy = createClientProxy(() -> "local", LocalService.class, tracingEnabled()); + assertEquals("local", serviceProxy.call()); + spans.assertKind("LocalService.call", SpanKind.CLIENT); + } + } + + @Test + public void testImplClassAnnotationOverridesKind() { + try (TestSpanCollector spans = TestSpanCollector.install("TestService")) { + LocalService serviceProxy = createClientProxy(new LocalServiceImpl(), LocalService.class, tracingEnabled()); + assertEquals("local", serviceProxy.call()); + spans.assertKind("LocalService.call", SpanKind.INTERNAL); + } + } + + @Test + public void testCreateProxyKinds() { + try (TestSpanCollector spans = TestSpanCollector.install("TestService")) { + createProxy(new ServiceImpl(), Service.class, tracingEnabled()).normalMethod(); + spans.assertKind("Service.normalMethod", SpanKind.INTERNAL); + } + try (TestSpanCollector spans = TestSpanCollector.install("TestService")) { + createClientProxy(new ServiceImpl(), Service.class, tracingEnabled()).normalMethod(); + spans.assertKind("Service.normalMethod", SpanKind.CLIENT); + } + } + + /** + * Interface with no tracing annotations. + */ + public interface LocalService { + String call(); + } + + /** + * Implementation whose methods are all in-process. + */ + @InternalSpanKind + public static class LocalServiceImpl implements LocalService { + @Override + public String call() { + return "local"; + } + } + } diff --git a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/tracing/TestTracingInitModes.java b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/tracing/TestTracingInitModes.java index 5ea02732c68..0e1a7188716 100644 --- a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/tracing/TestTracingInitModes.java +++ b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/tracing/TestTracingInitModes.java @@ -20,11 +20,14 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; import io.opentelemetry.api.GlobalOpenTelemetry; import io.opentelemetry.api.OpenTelemetry; import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanKind; import io.opentelemetry.context.Scope; import io.opentelemetry.sdk.OpenTelemetrySdk; import io.opentelemetry.sdk.trace.SdkTracerProvider; @@ -32,6 +35,8 @@ import io.opentelemetry.sdk.trace.export.SpanExporter; import org.apache.hadoop.hdds.conf.InMemoryConfigurationForTesting; import org.apache.hadoop.hdds.conf.MutableConfigurationSource; +import org.apache.hadoop.hdds.tracing.TestTraceAllMethod.Service; +import org.apache.hadoop.hdds.tracing.TestTraceAllMethod.ServiceImpl; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -166,6 +171,29 @@ public void testApplicationAwareAdoptsGlobalTracer() { provider.shutdown(); } + /** app-aware: a SpanKind proxy made with no current span still traces calls made under an app span. */ + @Test + public void testApplicationAwareSpanKindProxyCreatedAtStartup() { + installNoExportGlobalOpenTelemetry(); + MutableConfigurationSource conf = config(false, true); + TracingUtil.initTracing("app-aware-proxy", conf); + ServiceImpl delegate = new ServiceImpl(); + assertSame(delegate, TracingUtil.createProxy(delegate, Service.class, conf)); + Service proxy = TracingUtil.createProxy(delegate, Service.class, SpanKind.INTERNAL); + assertNotSame(delegate, proxy); + + proxy.normalMethod(); + assertFalse(delegate.wasSpanActive(), "Application-aware mode must NOT manufacture a root span"); + + Span appSpan = GlobalOpenTelemetry.getTracer("app").spanBuilder("app-root").startSpan(); + try (Scope ignored = appSpan.makeCurrent()) { + proxy.normalMethod(); + assertTrue(delegate.wasSpanActive(), "Proxy call under an app span should create a child span"); + } finally { + appSpan.end(); + } + } + /** Both flags false: tracing is off. */ @Test public void testOffModeIsInactive() { diff --git a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/tracing/TestTracingUtil.java b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/tracing/TestTracingUtil.java index 5b3e805700a..ad1df019136 100644 --- a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/tracing/TestTracingUtil.java +++ b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/tracing/TestTracingUtil.java @@ -31,16 +31,25 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.common.AttributeKey; import io.opentelemetry.api.trace.Span; import io.opentelemetry.api.trace.SpanContext; import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.StatusCode; import io.opentelemetry.api.trace.TraceFlags; import io.opentelemetry.api.trace.TraceState; import io.opentelemetry.context.Context; import io.opentelemetry.context.ContextStorage; import io.opentelemetry.context.Scope; +import io.opentelemetry.sdk.OpenTelemetrySdk; import io.opentelemetry.sdk.trace.ReadableSpan; +import io.opentelemetry.sdk.trace.SdkTracerProvider; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor; +import io.opentelemetry.sdk.trace.export.SpanExporter; import java.io.IOException; +import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import org.apache.hadoop.hdds.conf.InMemoryConfigurationForTesting; @@ -48,6 +57,7 @@ import org.apache.hadoop.hdds.tracing.TestTraceAllMethod.Service; import org.apache.hadoop.hdds.tracing.TestTraceAllMethod.ServiceImpl; import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.slf4j.MDC; @@ -73,6 +83,22 @@ private static TracingUtil.MdcContextStorage mdcContextStorage() { return new TracingUtil.MdcContextStorage(ContextStorage.defaultStorage()); } + /** Puts a real GlobalOpenTelemetry in place with no exporter, so tests stay offline and shut down fast. */ + @BeforeEach + public void setUp() { + TracingUtil.shutdownTracing(); + GlobalOpenTelemetry.resetForTest(); + SdkTracerProvider provider = + SdkTracerProvider.builder().addSpanProcessor(SimpleSpanProcessor.create(SpanExporter.noop())).build(); + GlobalOpenTelemetry.set(OpenTelemetrySdk.builder().setTracerProvider(provider).build()); + } + + @AfterEach + public void cleanup() { + TracingUtil.shutdownTracing(); + GlobalOpenTelemetry.resetForTest(); + } + private static MutableConfigurationSource tracingEnabled() { MutableConfigurationSource config = new InMemoryConfigurationForTesting(); config.setBoolean("ozone.tracing.enabled", true); @@ -110,6 +136,45 @@ public void testDefaultMethod() { assertEquals("Hello default", subject.defaultMethod()); } + @Test + public void testSpanKindProxyReturnsDelegateWhenTracingOff() { + ServiceImpl delegate = new ServiceImpl(); + assertSame(delegate, createProxy(delegate, Service.class, SpanKind.INTERNAL)); + } + + @Test + public void testSpanKindProxyCreatesRootSpanWithoutParent() { + try (TestSpanCollector collector = TestSpanCollector.install("test-service")) { + Service proxy = createProxy(new ServiceImpl(), Service.class, SpanKind.INTERNAL); + assertEquals("normal", proxy.normalMethod()); + collector.assertKind("ServiceImpl.normalMethod", SpanKind.INTERNAL); + assertFalse(finishedSpan(collector, "ServiceImpl.normalMethod").getParentSpanContext().isValid()); + } + } + + @Test + public void testSpanKindProxyCreatesChildSpans() { + try (TestSpanCollector collector = TestSpanCollector.install("test-service")) { + Service proxy = createProxy(new ServiceImpl(), Service.class, SpanKind.CLIENT); + Span parent = TracingUtil.createSpan("parent"); + try (Scope ignored = parent.makeCurrent()) { + assertEquals("normal", proxy.normalMethod()); + assertEquals("skipped", proxy.skippedMethod()); + assertThrows(IOException.class, proxy::tracedThrowingMethod); + } finally { + parent.end(); + } + + collector.assertParent("ServiceImpl.normalMethod", "parent"); + collector.assertKind("ServiceImpl.normalMethod", SpanKind.CLIENT); + assertTrue(collector.named("ServiceImpl.skippedMethod").isEmpty()); + collector.assertParent("ServiceImpl.tracedThrowingMethod", "parent"); + collector.assertError("ServiceImpl.tracedThrowingMethod", "Failed"); + collector.assertExceptionRecorded("ServiceImpl.tracedThrowingMethod", IOException.class); + collector.assertNotError("parent"); + } + } + @Test public void testInitTracing() { TracingUtil.initTracing("testInitTracing", tracingEnabled()); @@ -194,6 +259,33 @@ public void testExecuteAsChildSpanPropagatesException() throws Exception { assertEquals("expected", thrown.getMessage()); } + @Test + public void testExecuteInSpanRecordsException() { + try (TestSpanCollector collector = TestSpanCollector.install("test")) { + assertThrows(IOException.class, () -> TracingUtil.executeInNewSpan("supplier", () -> { + throw new IOException("supplier failed"); + })); + Span parent = TracingUtil.createSpan("parent"); + try (Scope ignored = parent.makeCurrent()) { + String parentContext = TracingUtil.exportCurrentSpan(); + assertThrows(IllegalStateException.class, () -> TracingUtil.executeAsChildSpan("runnable", parentContext, + () -> { + throw new IllegalStateException("runnable failed"); + })); + } finally { + parent.end(); + } + + collector.assertError("supplier", "supplier failed"); + collector.assertExceptionRecorded("supplier", IOException.class); + assertEquals("Failed with exception: supplier failed", firstEventName(collector, "supplier")); + collector.assertError("runnable", "runnable failed"); + collector.assertExceptionRecorded("runnable", IllegalStateException.class); + assertEquals("Failed with exception: runnable failed", firstEventName(collector, "runnable")); + collector.assertNotError("parent"); + } + } + @Test public void testTextExtractorAsTextMapGetter() { TracingUtil.TextExtractor getter = new TracingUtil.TextExtractor(); @@ -334,4 +426,160 @@ public void testMdcContextStorageOnOtherThread() throws Exception { executor.shutdownNow(); } } + + @Test + public void testCreateSpanIsChildOfCurrentSpan() { + try (TestSpanCollector collector = TestSpanCollector.install("test")) { + Span parent = TracingUtil.createSpan("parent"); + try (Scope ignored = parent.makeCurrent()) { + Span child = TracingUtil.createSpan("child", SpanKind.CLIENT); + child.end(); + } finally { + parent.end(); + } + + SpanData parentData = finishedSpan(collector, "parent"); + SpanData childData = finishedSpan(collector, "child"); + assertEquals(SpanKind.INTERNAL, parentData.getKind()); + assertEquals(SpanKind.CLIENT, childData.getKind()); + assertEquals(parentData.getTraceId(), childData.getTraceId()); + assertEquals(parentData.getSpanId(), childData.getParentSpanId()); + } + } + + @Test + public void testCreateSpanWithoutParentStartsRootWhenEnabled() { + try (TestSpanCollector collector = TestSpanCollector.install("test")) { + Span span = TracingUtil.createSpan("root"); + assertTrue(span.getSpanContext().isValid()); + span.end(); + assertFalse(finishedSpan(collector, "root").getParentSpanContext().isValid()); + } + } + + @Test + public void testRecordExceptionSetsErrorAndAddsEvent() { + try (TestSpanCollector collector = TestSpanCollector.install("test")) { + Span span = TracingUtil.createSpan("failing"); + try (Scope ignored = span.makeCurrent()) { + TracingUtil.recordExceptionInCurrentSpan(new IOException("disk failed")); + } finally { + span.end(); + } + + SpanData data = finishedSpan(collector, "failing"); + assertEquals(StatusCode.ERROR, data.getStatus().getStatusCode()); + assertEquals("disk failed", data.getStatus().getDescription()); + assertEquals(1, data.getEvents().size()); + assertEquals("exception", data.getEvents().get(0).getName()); + assertEquals(IOException.class.getName(), + data.getEvents().get(0).getAttributes().get(AttributeKey.stringKey("exception.type"))); + } + } + + @Test + public void testRecordExceptionInCatchMarksSpanNotParent() { + try (TestSpanCollector collector = TestSpanCollector.install("test")) { + Span parent = TracingUtil.createSpan("parent"); + try (Scope ignored = parent.makeCurrent()) { + Span span = TracingUtil.createSpan("failing"); + try (Scope ignored2 = span.makeCurrent()) { + throw new IOException("disk failed"); + } catch (IOException e) { + TracingUtil.recordException(span, e); + } finally { + span.end(); + } + } finally { + parent.end(); + } + + collector.assertError("failing", "disk failed"); + collector.assertExceptionRecorded("failing", IOException.class); + assertEquals(StatusCode.UNSET, finishedSpan(collector, "parent").getStatus().getStatusCode()); + assertTrue(finishedSpan(collector, "parent").getEvents().isEmpty()); + } + } + + @Test + public void testRecordExceptionWithoutMessageSetsError() { + try (TestSpanCollector collector = TestSpanCollector.install("test")) { + Span span = TracingUtil.createSpan("failing"); + try (Scope ignored = span.makeCurrent()) { + TracingUtil.recordExceptionInCurrentSpan(new IllegalStateException()); + } finally { + span.end(); + } + SpanData data = finishedSpan(collector, "failing"); + assertEquals(StatusCode.ERROR, data.getStatus().getStatusCode()); + assertEquals("", data.getStatus().getDescription()); + } + } + + @Test + public void testRecordExceptionIgnoresNull() { + try (TestSpanCollector collector = TestSpanCollector.install("test")) { + Span span = TracingUtil.createSpan("ok"); + try (Scope ignored = span.makeCurrent()) { + TracingUtil.recordExceptionInCurrentSpan(null); + } finally { + span.end(); + } + assertEquals(StatusCode.UNSET, finishedSpan(collector, "ok").getStatus().getStatusCode()); + } + } + + @Test + public void testCurrentSpanHelpersSetAttributesAndEvents() { + try (TestSpanCollector collector = TestSpanCollector.install("test")) { + Span span = TracingUtil.createSpan("request"); + try (Scope ignored = span.makeCurrent()) { + TracingUtil.addAttributesToCurrentSpan("volume", "vol1"); + TracingUtil.addAttributesToCurrentSpan("bucket", (String) null); + TracingUtil.addAttributesToCurrentSpan("dataSize", 1024L); + TracingUtil.addAttributesToCurrentSpan("om.success", false); + TracingUtil.addEventToCurrentSpan("client.datanode.failover", "datanode", "dn1"); + TracingUtil.addEventToCurrentSpan("scm.allocate.failed", "reason", null); + } finally { + span.end(); + } + + SpanData data = finishedSpan(collector, "request"); + assertEquals("vol1", data.getAttributes().get(AttributeKey.stringKey("volume"))); + assertNull(data.getAttributes().get(AttributeKey.stringKey("bucket"))); + assertEquals(1024L, data.getAttributes().get(AttributeKey.longKey("dataSize"))); + assertEquals(false, data.getAttributes().get(AttributeKey.booleanKey("om.success"))); + assertEquals(StatusCode.UNSET, data.getStatus().getStatusCode()); + + assertEquals(2, data.getEvents().size()); + assertEquals("client.datanode.failover", data.getEvents().get(0).getName()); + assertEquals("dn1", data.getEvents().get(0).getAttributes().get(AttributeKey.stringKey("datanode"))); + assertEquals("scm.allocate.failed", data.getEvents().get(1).getName()); + assertTrue(data.getEvents().get(1).getAttributes().isEmpty()); + } + } + + @Test + public void testCurrentSpanHelpersWithoutCurrentSpanAreNoOp() { + try (TestSpanCollector collector = TestSpanCollector.install("test")) { + assertFalse(Span.current().getSpanContext().isValid()); + TracingUtil.addAttributesToCurrentSpan("volume", "vol1"); + TracingUtil.addAttributesToCurrentSpan("dataSize", 1L); + TracingUtil.addAttributesToCurrentSpan("om.success", true); + TracingUtil.addEventToCurrentSpan("event", "k", "v"); + TracingUtil.recordExceptionInCurrentSpan(new IOException("ignored")); + assertTrue(collector.getSpans().isEmpty()); + } + } + + /** @return the only collected span named {@code name}; fails the test if there is not exactly one. */ + private static SpanData finishedSpan(TestSpanCollector collector, String name) { + List spans = collector.named(name); + assertEquals(1, spans.size(), () -> "spans named " + name + ": " + collector.getSpans()); + return spans.get(0).toSpanData(); + } + + private static String firstEventName(TestSpanCollector collector, String name) { + return finishedSpan(collector, name).getEvents().get(0).getName(); + } } diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/server/OzoneProtocolMessageDispatcher.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/server/OzoneProtocolMessageDispatcher.java index 29fc54642eb..8d5f73d6e05 100644 --- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/server/OzoneProtocolMessageDispatcher.java +++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/server/OzoneProtocolMessageDispatcher.java @@ -101,6 +101,9 @@ public RESPONSE processRequest( } return response; + } catch (Throwable t) { + TracingUtil.recordException(span, t); + throw t; } finally { span.end(); } diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java index 658b79d5621..7ec4b062297 100644 --- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java +++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java @@ -145,7 +145,7 @@ public static ScmBlockLocationProtocol getScmBlockClient( ScmBlockLocationProtocolClientSideTranslatorPB scmBlockLocationClient = new ScmBlockLocationProtocolClientSideTranslatorPB(proxyProvider, conf); return TracingUtil - .createProxy(scmBlockLocationClient, ScmBlockLocationProtocol.class, + .createClientProxy(scmBlockLocationClient, ScmBlockLocationProtocol.class, conf); } @@ -163,7 +163,7 @@ public static StorageContainerLocationInternalInterface getScmContainerClient( ConfigurationSource conf, SCMContainerLocationFailoverProxyProvider proxyProvider) { StorageContainerLocationInternalInterface scmContainerClient = - TracingUtil.createProxy( + TracingUtil.createClientProxy( new StorageContainerLocationProtocolClientSideTranslatorPB( proxyProvider), StorageContainerLocationInternalInterface.class, conf); return scmContainerClient; @@ -176,7 +176,7 @@ public static StorageContainerLocationInternalInterface getScmContainerClient( new SCMContainerLocationFailoverProxyProvider(conf, userGroupInformation); StorageContainerLocationInternalInterface scmContainerClient = - TracingUtil.createProxy( + TracingUtil.createClientProxy( new StorageContainerLocationProtocolClientSideTranslatorPB( proxyProvider), StorageContainerLocationInternalInterface.class, conf); return scmContainerClient; @@ -192,7 +192,7 @@ public static StorageContainerLocationInternalInterface getScmContainerClientFor SCMContainerLocationFailoverProxyProvider proxyProvider = new SCMContainerLocationFailoverProxyProvider(conf, ugi); StorageContainerLocationInternalInterface scmContainerClient = - TracingUtil.createProxy( + TracingUtil.createClientProxy( new StorageContainerLocationProtocolClientSideTranslatorPB( proxyProvider, targetScmNode), StorageContainerLocationInternalInterface.class, conf); return scmContainerClient; diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HddsServerUtil.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HddsServerUtil.java index 1d0ce81f004..b904f8f2a1a 100644 --- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HddsServerUtil.java +++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HddsServerUtil.java @@ -658,7 +658,7 @@ public static SCMSecurityProtocolClientSideTranslatorPB getScmSecurityClient( SCMSecurityProtocolClientSideTranslatorPB scmSecurityClient = new SCMSecurityProtocolClientSideTranslatorPB( new SCMSecurityProtocolFailoverProxyProvider(conf, ugi)); - return TracingUtil.createProxy(scmSecurityClient, + return TracingUtil.createClientProxy(scmSecurityClient, SCMSecurityProtocolClientSideTranslatorPB.class, conf); } @@ -678,7 +678,7 @@ public static SecretKeyProtocolScm getSecretKeyClientForSCM( UserGroupInformation.getCurrentUser(), SecretKeyProtocolScmPB.class), SecretKeyProtocolScmPB.class); - return TracingUtil.createProxy(scmSecretClient, + return TracingUtil.createClientProxy(scmSecretClient, SecretKeyProtocolScm.class, conf); } diff --git a/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/server/TestOzoneProtocolMessageDispatcher.java b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/server/TestOzoneProtocolMessageDispatcher.java new file mode 100644 index 00000000000..10e5de46d60 --- /dev/null +++ b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/server/TestOzoneProtocolMessageDispatcher.java @@ -0,0 +1,76 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.hdds.server; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import com.google.protobuf.ServiceException; +import io.opentelemetry.api.trace.SpanKind; +import org.apache.hadoop.hdds.tracing.TestSpanCollector; +import org.apache.hadoop.hdds.utils.ProtocolMessageMetrics; +import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; + +/** + * Tests the SERVER span created by {@link OzoneProtocolMessageDispatcher}. + */ +public class TestOzoneProtocolMessageDispatcher { + + private final OzoneProtocolMessageDispatcher dispatcher = + new OzoneProtocolMessageDispatcher<>("test", ProtocolMessageMetrics.create("test", "test", TestType.class), + LoggerFactory.getLogger(TestOzoneProtocolMessageDispatcher.class)); + + @Test + public void testSuccessfulRequestIsNotError() throws ServiceException { + try (TestSpanCollector spans = TestSpanCollector.install("test")) { + assertEquals("response", dispatcher.processRequest("request", request -> "response", TestType.SUCCEED, null)); + spans.assertKind(TestType.SUCCEED.name(), SpanKind.SERVER); + spans.assertNotError(TestType.SUCCEED.name()); + } + } + + @Test + public void testThrownExceptionMarksSpanError() { + try (TestSpanCollector spans = TestSpanCollector.install("test")) { + ServiceException ex = assertThrows(ServiceException.class, () -> dispatcher.processRequest("request", + request -> { + throw new ServiceException("request failed"); + }, TestType.FAIL, null)); + assertEquals("request failed", ex.getMessage()); + spans.assertKind(TestType.FAIL.name(), SpanKind.SERVER); + spans.assertError(TestType.FAIL.name(), "request failed"); + spans.assertExceptionRecorded(TestType.FAIL.name(), ServiceException.class); + } + } + + @Test + public void testRuntimeExceptionMarksSpanError() { + try (TestSpanCollector spans = TestSpanCollector.install("test")) { + assertThrows(IllegalStateException.class, () -> dispatcher.processRequest("request", request -> { + throw new IllegalStateException("bug"); + }, TestType.FAIL, null)); + spans.assertError(TestType.FAIL.name(), "bug"); + spans.assertExceptionRecorded(TestType.FAIL.name(), IllegalStateException.class); + } + } + + private enum TestType { + SUCCEED, FAIL + } +} diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java index 6ce6966934c..6fdffa4a02a 100644 --- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java +++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java @@ -276,7 +276,7 @@ public RpcClient(ConfigurationSource conf, String omServiceId) ozoneManagerProtocolClientSideTranslatorPB = new OzoneManagerProtocolClientSideTranslatorPB(omTransport, clientId.toString()); - this.ozoneManagerClient = TracingUtil.createProxy( + this.ozoneManagerClient = TracingUtil.createClientProxy( ozoneManagerProtocolClientSideTranslatorPB, OzoneManagerClientProtocol.class, conf); dtService = omTransport.getDelegationTokenService(); diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestFailoverWithSCMHA.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestFailoverWithSCMHA.java index c98c7d767c3..1fb8093424e 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestFailoverWithSCMHA.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestFailoverWithSCMHA.java @@ -119,7 +119,7 @@ public void testFailover() throws Exception { .setLogLevel(SCMBlockLocationFailoverProxyProvider.class, Level.DEBUG); LogCapturer logCapture = LogCapturer.captureLogs(SCMBlockLocationFailoverProxyProvider.class); ScmBlockLocationProtocol scmBlockLocationProtocol = TracingUtil - .createProxy(scmBlockLocationClient, ScmBlockLocationProtocol.class, + .createClientProxy(scmBlockLocationClient, ScmBlockLocationProtocol.class, conf); scmBlockLocationProtocol.getScmInfo(); assertThat(logCapture.getOutput()) @@ -132,7 +132,7 @@ public void testFailover() throws Exception { logCapture = LogCapturer.captureLogs(SCMContainerLocationFailoverProxyProvider.class); proxyProvider.changeCurrentProxy(scm.getSCMNodeId()); StorageContainerLocationInternalInterface scmContainerClient = - TracingUtil.createProxy( + TracingUtil.createClientProxy( new StorageContainerLocationProtocolClientSideTranslatorPB( proxyProvider), StorageContainerLocationInternalInterface.class, conf);