HPCC-586 WsClient OTEL tracing

- Injects trace context in traceparent http header - Integrates autoconfigure otel SDK - Adds manual WsWUClient.ping span - Adds manual WsWUClientTest.ping span - Adds manual span around getHPCCver and getHPCCContainerizedMode Signed-off-by: Rodrigo Pastrana <[email protected]>
hpcc-systems · May 31, 2024 · ad53e15 · ad53e15
1 parent 45f0071
commit ad53e15
Show file tree

Hide file tree

Showing 6 changed files with 242 additions and 49 deletions.
diff --git a/pom.xml b/pom.xml
@@ -63,6 +63,8 @@
     <groups>org.hpccsystems.commons.annotations.BaseTests</groups>
     <codehaus.template.version>1.0.0</codehaus.template.version>
     <project.benchmarking>false</project.benchmarking>
+    <opentelemetry.bom.version>1.38.0</opentelemetry.bom.version>
+    <opentelemetry.semconv.version>1.25.0-alpha</opentelemetry.semconv.version>
   </properties>
 
   <scm>
@@ -98,8 +100,48 @@
       </snapshots>
     </repository>
   </repositories>
-
+  <dependencyManagement>
+    <dependencies>
+      <dependency>
+        <groupId>io.opentelemetry</groupId>
+        <artifactId>opentelemetry-bom</artifactId>
+        <version>${opentelemetry.bom.version}</version>
+        <type>pom</type>
+        <scope>import</scope>
+      </dependency>
+    </dependencies>
+  </dependencyManagement>
   <dependencies>
+    <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-sdk</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-exporter-logging</artifactId>
+        </dependency>
+        <dependency>
+          <groupId>io.opentelemetry</groupId>
+          <artifactId>opentelemetry-sdk-extension-autoconfigure</artifactId>
+        </dependency>
+        <dependency>
+          <groupId>io.opentelemetry</groupId>
+          <artifactId>opentelemetry-sdk-extension-autoconfigure-spi</artifactId>
+        </dependency>
+        <dependency>
+          <groupId>io.opentelemetry</groupId>
+          <artifactId>opentelemetry-exporter-otlp</artifactId>
+        </dependency>
+        <dependency>
+          <!-- Not managed by opentelemetry-bom -->
+          <groupId>io.opentelemetry.semconv</groupId>
+          <artifactId>opentelemetry-semconv</artifactId>
+          <version>1.25.0-alpha</version>
+        </dependency>
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>

diff --git a/wsclient/src/main/java/org/hpccsystems/ws/client/BaseHPCCWsClient.java b/wsclient/src/main/java/org/hpccsystems/ws/client/BaseHPCCWsClient.java
@@ -4,6 +4,8 @@
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
 
 import javax.xml.parsers.DocumentBuilder;
 import javax.xml.parsers.ParserConfigurationException;
@@ -40,13 +42,29 @@
 import org.w3c.dom.Node;
 import org.w3c.dom.NodeList;
 
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanBuilder;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+import io.opentelemetry.context.propagation.TextMapSetter;
+import io.opentelemetry.semconv.HttpAttributes;
+import io.opentelemetry.semconv.ServerAttributes;
+
 /**
- * Defines functionality common to all HPCC Systmes web service clients.
+ * Defines functionality common to all HPCC Systems web service clients.
  *
  * Typically implemented by specialized HPCC Web service clients.
  */
 public abstract class BaseHPCCWsClient extends DataSingleton
 {
+    public static final String INSTRUMENTED_LIBRARY_NAME = "WsClient";
+    private static OpenTelemetry globalOTel = null;
     /** Constant <code>log</code> */
     protected static final Logger log                    = LogManager.getLogger(BaseHPCCWsClient.class);
     /** Constant <code>DEAFULTECLWATCHPORT="8010"</code> */
@@ -164,6 +182,79 @@ private String getTargetHPCCBuildVersionString() throws Exception
 
     }
 
+    public SpanBuilder getWsClientSpanBuilder(String spanName)
+    {
+        SpanBuilder spanBuilder = getWsClientTracer().spanBuilder(spanName)
+          .setAttribute(ServerAttributes.SERVER_ADDRESS, wsconn.getHost())
+          .setAttribute(ServerAttributes.SERVER_PORT, Long.getLong(wsconn.getPort()))
+          .setAttribute(HttpAttributes.HTTP_REQUEST_METHOD, HttpAttributes.HttpRequestMethodValues.GET)
+          .setSpanKind(SpanKind.CLIENT);
+
+        return spanBuilder;
+    }
+
+    static public void injectCurrentSpanTraceParentHeader(Stub clientStub)
+    {
+        if (clientStub != null)
+        {
+            injectCurrentSpanTraceParentHeader(clientStub._getServiceClient().getOptions());
+        }
+    }
+
+    static public void injectCurrentSpanTraceParentHeader(Options options)
+    {
+        if (options != null)
+        {
+            W3CTraceContextPropagator.getInstance().inject(Context.current(), options, Options::setProperty);
+        }
+    }
+
+    static public String getCurrentSpanTraceParentHeader()
+    {
+        String traceparent = null;
+        Span currentSpan = Span.current();
+        if (currentSpan != null && currentSpan.getSpanContext().isValid())
+        {
+            Map<String, String> carrier = new HashMap<>();
+            TextMapSetter<Map<String, String>> setter = Map::put;
+            W3CTraceContextPropagator.getInstance().inject(Context.current(), carrier, setter);
+
+            traceparent = carrier.getOrDefault("traceparent", "00-" + currentSpan.getSpanContext().getTraceId() + "-" + currentSpan.getSpanContext().getSpanId() + "-00");
+            carrier.clear();
+        }
+
+        return traceparent;
+    }
+
+    /**
+     * Performs all Otel initialization
+     */
+    private void initOTel()
+    {
+        /*
+         * If using the OpenTelemetry SDK, you may want to instantiate the OpenTelemetry toprovide configuration, for example of Resource or Sampler. See OpenTelemetrySdk and OpenTelemetrySdk.builder for information on how to construct theSDK's OpenTelemetry implementation. 
+         * WARNING: Due to the inherent complications around initialization order involving this classand its single global instance, we strongly recommend *not* using GlobalOpenTelemetry unless youhave a use-case that absolutely requires it. Please favor using instances of OpenTelemetrywherever possible. 
+         *  If you are using the OpenTelemetry javaagent, it is generally best to only callGlobalOpenTelemetry.get() once, and then pass the resulting reference where you need to use it.
+        */
+
+        //autoconfigured telemetry should be configured via env vars or system attributes:
+        //https://github.com/open-telemetry/opentelemetry-java/blob/main/sdk-extensions/autoconfigure/README.md
+        //For Example:
+        //    -Dotel.traces.exporter=logging
+        //    -Dotel.metrics.exporter=none
+        //    -D.otel.logs.exporter=none
+        //    -Dotel.java.global-autoconfigure.enabled=true
+
+        globalOTel = GlobalOpenTelemetry.get();
+    }
+
+    public Tracer getWsClientTracer()
+    {
+        if (globalOTel == null)
+            initOTel();
+
+        return globalOTel.getTracer(INSTRUMENTED_LIBRARY_NAME);
+    }
     /**
      * All instances of HPCCWsXYZClient should utilize this init function
      * Attempts to establish the target HPCC build version and its container mode
@@ -175,36 +266,55 @@ private String getTargetHPCCBuildVersionString() throws Exception
      */
     protected boolean initBaseWsClient(Connection connection, boolean fetchVersionAndContainerMode)
     {
+        initOTel();
+
         boolean success = true;
         initErrMessage = "";
         setActiveConnectionInfo(connection);
 
         if (fetchVersionAndContainerMode)
         {
-            try
+            Span fetchHPCCVerSpan = getWsClientSpanBuilder("FetchHPCCVersion").startSpan();
+            try (Scope scope = fetchHPCCVerSpan.makeCurrent())
             {
-                targetHPCCBuildVersion = new Version(getTargetHPCCBuildVersionString());
+                try
+                {
+                    targetHPCCBuildVersion = new Version(getTargetHPCCBuildVersionString());
+                }
+                catch (Exception e)
+                {
+                    initErrMessage = "BaseHPCCWsClient: Could not stablish target HPCC bulid version, review all HPCC connection values";
+                    if (!e.getLocalizedMessage().isEmpty())
+                        initErrMessage = initErrMessage + "\n" + e.getLocalizedMessage();
+                    success = false;
+                }
             }
-            catch (Exception e)
+            finally
             {
-                initErrMessage = "BaseHPCCWsClient: Could not stablish target HPCC bulid version, review all HPCC connection values";
-                if (!e.getLocalizedMessage().isEmpty())
-                    initErrMessage = initErrMessage + "\n" + e.getLocalizedMessage();
-
-                success = false;
+                fetchHPCCVerSpan.setStatus(success ? StatusCode.OK : StatusCode.ERROR, initErrMessage);
+                fetchHPCCVerSpan.end();
             }
 
-            try
+            Span fetchHPCCContainerMode = getWsClientSpanBuilder("FetchHPCCContainerMode").startSpan();
+            try (Scope scope = fetchHPCCContainerMode.makeCurrent())
             {
-                targetsContainerizedHPCC = getTargetHPCCIsContainerized(wsconn);
+                try
+                {
+                    targetsContainerizedHPCC = getTargetHPCCIsContainerized(wsconn);
+                }
+                catch (Exception e)
+                {
+                    initErrMessage = initErrMessage + "\nBaseHPCCWsClient: Could not determine target HPCC Containerization mode, review all HPCC connection values";
+                    if (!e.getLocalizedMessage().isEmpty())
+                        initErrMessage = initErrMessage + "\n" + e.getLocalizedMessage();
+
+                    success = false;
+                }
             }
-            catch (Exception e)
+            finally
             {
-                initErrMessage = initErrMessage + "\nBaseHPCCWsClient: Could not determine target HPCC Containerization mode, review all HPCC connection values";
-                if (!e.getLocalizedMessage().isEmpty())
-                    initErrMessage = initErrMessage + "\n" + e.getLocalizedMessage();
-
-                success = false;
+                fetchHPCCContainerMode.setStatus(success ? StatusCode.OK : StatusCode.ERROR, initErrMessage);
+                fetchHPCCContainerMode.end();
             }
         }
         if (!initErrMessage.isEmpty())
@@ -401,7 +511,10 @@ public String getInitError()
     protected Stub verifyStub() throws Exception
     {
         if (stub != null)
+        {
+            injectCurrentSpanTraceParentHeader(stub);
             return stub;
+        }
         else
             throw new Exception("WS Client Stub not available." + (hasInitError() ? "\n" + initErrMessage : ""));
     }
@@ -682,6 +795,7 @@ protected void handleEspExceptions(ArrayOfEspExceptionWrapper exp, String messag
         if (message != null && !message.isEmpty()) exp.setWsClientMessage(message);
 
         log.error(exp.toString());
+
         throw exp;
     }
 

diff --git a/wsclient/src/main/java/org/hpccsystems/ws/client/HPCCWsClient.java b/wsclient/src/main/java/org/hpccsystems/ws/client/HPCCWsClient.java
@@ -19,7 +19,6 @@
 import org.hpccsystems.ws.client.utils.Utils;
 import org.hpccsystems.ws.client.wrappers.ArrayOfECLExceptionWrapper;
 import org.hpccsystems.ws.client.wrappers.gen.filespray.ProgressResponseWrapper;
-import org.hpccsystems.ws.client.wrappers.gen.wstopology.TpGroupQueryResponseWrapper;
 import org.hpccsystems.ws.client.wrappers.gen.wstopology.TpGroupWrapper;
 import org.hpccsystems.ws.client.wrappers.wsworkunits.WorkunitWrapper;
 

diff --git a/wsclient/src/main/java/org/hpccsystems/ws/client/HPCCWsWorkUnitsClient.java b/wsclient/src/main/java/org/hpccsystems/ws/client/HPCCWsWorkUnitsClient.java
@@ -100,6 +100,10 @@
 import org.hpccsystems.ws.client.wrappers.wsworkunits.WorkunitWrapper;
 import org.hpccsystems.ws.client.wrappers.wsworkunits.WsWorkunitsClientStubWrapper;
 
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.context.Scope;
+
 /**
  * Facilitates ECL WorkUnit related actions.
  *
@@ -303,7 +307,6 @@ public void fastWURefresh(WorkunitWrapper wu) throws Exception, ArrayOfEspExcept
         request.setCount(1);
 
         WUQueryResponse response = null;
-
         try
         {
             response = ((WsWorkunits) stub).wUQuery(request);
@@ -326,7 +329,8 @@ public void fastWURefresh(WorkunitWrapper wu) throws Exception, ArrayOfEspExcept
             {
                 ECLWorkunit[] eclWorkunit = response.getWorkunits().getECLWorkunit();
 
-                if (eclWorkunit != null && eclWorkunit.length == 1) wu.update(eclWorkunit[0]);
+                if (eclWorkunit != null && eclWorkunit.length == 1)
+                    wu.update(eclWorkunit[0]);
             }
 
             if (previousState != getStateID(wu))
@@ -2551,18 +2555,27 @@ public List<QueryResultWrapper> deleteQueries(Set<String> querynames, String clu
      */
     public boolean ping() throws Exception
     {
-        verifyStub();
-
-        Ping request = new Ping();
-
-        try
+        Span span = getWsClientSpanBuilder("WsWUClient_Ping").startSpan();
+        try (Scope scope = span.makeCurrent())
         {
-            ((WsWorkunitsStub) stub).ping(request);
+            verifyStub(); // must be called within span scope for proper context propagation
+
+            Ping request = new Ping();
+            try
+            {
+                ((WsWorkunitsStub) stub).ping(request);
+                span.setStatus(StatusCode.OK);
+            }
+            catch (Exception e)
+            {
+                span.recordException(e);
+                log.error(e.getLocalizedMessage());
+                return false;
+            }
         }
-        catch (Exception e)
+        finally
         {
-            log.error(e.getLocalizedMessage());
-            return false;
+            span.end();
         }
 
         return true;