Adding destination to Messaging transactions and spans (#906)

Author: eyalkoren

CHANGELOG.md

@@ -15,6 +15,7 @@
  * Logs message when `transaction_max_spans` has been exceeded (#849)
  * Report the number of affected rows by a SQL statement (UPDATE,DELETE,INSERT) in 'affected_rows' span attribute (#707)
  * Add [`@Traced`](https://www.elastic.co/guide/en/apm/agent/java/master/public-api.html#api-traced) annotation which either creates a span or a transaction, depending on the context
+ * Report JMS destination as a span/transaction context field (#906)
 
 ## Bug Fixes
  * JMS creates polling transactions even when the API invocations return without a message

apm-agent-core/src/main/java/co/elastic/apm/agent/impl/context/AbstractContext.java

@@ -40,6 +40,11 @@ public abstract class AbstractContext implements Recyclable {
  */
  private final Map<String, Object> labels = new ConcurrentHashMap<>();
 
+ /**
+ * An object containing contextual data for Messages (incoming in case of transactions or outgoing in case of spans)
+ */
+ private final Message message = new Message();
+
  public Iterator<? extends Map.Entry<String, ?>> getLabelIterator() {
  return labels.entrySet().iterator();
  }
@@ -68,16 +73,22 @@ public boolean hasLabels() {
  return !labels.isEmpty();
  }
 
+ public Message getMessage() {
+ return message;
+ }
+
  @Override
  public void resetState() {
  labels.clear();
+ message.resetState();
  }
 
  public boolean hasContent() {
- return !labels.isEmpty();
+ return !labels.isEmpty() || message.hasContent();
  }
 
  public void copyFrom(AbstractContext other) {
  labels.putAll(other.labels);
+ message.copyFrom(other.message);
  }
 }

apm-agent-core/src/main/java/co/elastic/apm/agent/impl/transaction/Db.java renamed to apm-agent-core/src/main/java/co/elastic/apm/agent/impl/context/Db.java

@@ -23,7 +23,7 @@
  * #L%
  */
 
-package co.elastic.apm.agent.impl.transaction;
+package co.elastic.apm.agent.impl.context;
 
 import co.elastic.apm.agent.objectpool.Allocator;
 import co.elastic.apm.agent.objectpool.ObjectPool;

apm-agent-core/src/main/java/co/elastic/apm/agent/impl/transaction/Http.java renamed to apm-agent-core/src/main/java/co/elastic/apm/agent/impl/context/Http.java

@@ -22,7 +22,7 @@
  * under the License.
  * #L%
  */
-package co.elastic.apm.agent.impl.transaction;
+package co.elastic.apm.agent.impl.context;
 
 import co.elastic.apm.agent.objectpool.Recyclable;
 

apm-agent-core/src/main/java/co/elastic/apm/agent/impl/context/Message.java

@@ -0,0 +1,73 @@
+/*-
+ * #%L
+ * Elastic APM Java agent
+ * %%
+ * Copyright (C) 2018 - 2019 Elastic and contributors
+ * %%
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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.
+ * #L%
+ */
+package co.elastic.apm.agent.impl.context;
+
+import co.elastic.apm.agent.objectpool.Recyclable;
+
+import javax.annotation.Nullable;
+
+public class Message implements Recyclable {
+
+ @Nullable
+ private String queueName;
+
+ @Nullable
+ private String topicName;
+
+ @Nullable
+ public String getQueueName() {
+ return queueName;
+ }
+
+ public Message withQueue(String queueName) {
+ this.queueName = queueName;
+ return this;
+ }
+
+ @Nullable
+ public String getTopicName() {
+ return topicName;
+ }
+
+ public Message withTopic(String topicName) {
+ this.topicName = topicName;
+ return this;
+ }
+
+ public boolean hasContent() {
+ return queueName != null || topicName != null;
+ }
+
+ @Override
+ public void resetState() {
+ queueName = null;
+ topicName = null;
+ }
+
+ public void copyFrom(Message other) {
+ this.queueName = other.getQueueName();
+ this.topicName = other.getTopicName();
+ }
+}

apm-agent-core/src/main/java/co/elastic/apm/agent/impl/context/SpanContext.java

@@ -25,9 +25,6 @@
 
 package co.elastic.apm.agent.impl.context;
 
-import co.elastic.apm.agent.impl.transaction.Db;
-import co.elastic.apm.agent.impl.transaction.Http;
-
 
 /**
  * Any other arbitrary data captured by the agent, optionally provided by the user

apm-agent-core/src/main/java/co/elastic/apm/agent/report/serialize/DslJsonSerializer.java

@@ -26,6 +26,7 @@
 
 import co.elastic.apm.agent.impl.MetaData;
 import co.elastic.apm.agent.impl.context.AbstractContext;
+import co.elastic.apm.agent.impl.context.Message;
 import co.elastic.apm.agent.impl.context.Request;
 import co.elastic.apm.agent.impl.context.Response;
 import co.elastic.apm.agent.impl.context.Socket;
@@ -46,8 +47,8 @@
 import co.elastic.apm.agent.impl.payload.SystemInfo;
 import co.elastic.apm.agent.impl.payload.TransactionPayload;
 import co.elastic.apm.agent.impl.stacktrace.StacktraceConfiguration;
-import co.elastic.apm.agent.impl.transaction.Db;
-import co.elastic.apm.agent.impl.transaction.Http;
+import co.elastic.apm.agent.impl.context.Db;
+import co.elastic.apm.agent.impl.context.Http;
 import co.elastic.apm.agent.impl.transaction.Id;
 import co.elastic.apm.agent.impl.transaction.Span;
 import co.elastic.apm.agent.impl.transaction.SpanCount;
@@ -723,6 +724,7 @@ private void serializeSpanContext(SpanContext context, TraceContext traceContext
  jw.writeByte(OBJECT_START);
 
  serializeServiceName(traceContext);
+ serializeMessageContext(context.getMessage());
  serializeDbContext(context.getDb());
  serializeHttpContext(context.getHttp());
 
@@ -733,6 +735,26 @@ private void serializeSpanContext(SpanContext context, TraceContext traceContext
  jw.writeByte(COMMA);
  }
 
+ private void serializeMessageContext(final Message message) {
+ if (message.hasContent()) {
+ writeFieldName("message");
+ jw.writeByte(OBJECT_START);
+ if (message.getTopicName() != null) {
+ writeFieldName("topic");
+ jw.writeByte(OBJECT_START);
+ writeLastField("name", message.getTopicName());
+ jw.writeByte(OBJECT_END);
+ } else {
+ writeFieldName("queue");
+ jw.writeByte(OBJECT_START);
+ writeLastField("name", message.getQueueName());
+ jw.writeByte(OBJECT_END);
+ }
+ jw.writeByte(OBJECT_END);
+ jw.writeByte(COMMA);
+ }
+ }
+
  private void serializeDbContext(final Db db) {
  if (db.hasContent()) {
  writeFieldName("db");
@@ -798,6 +820,7 @@ private void serializeContext(final TransactionContext context, TraceContext tra
  }
  serializeRequest(context.getRequest());
  serializeResponse(context.getResponse());
+ serializeMessageContext(context.getMessage());
  if (context.hasCustom()) {
  writeFieldName("custom");
  serializeStringKeyScalarValueMap(context.getCustomIterator(), replaceBuilder, jw, true, true);

apm-agent-core/src/test/java/co/elastic/apm/agent/report/serialize/DslJsonSerializerTest.java

@@ -244,6 +244,11 @@ void testSpanTypeSerialization() {
  span.withType("template.jsf.render.view");
  JsonNode spanJson = readJsonString(serializer.toJsonString(span));
  assertThat(spanJson.get("type").textValue()).isEqualTo("template_jsf_render_view");
+ JsonNode context = spanJson.get("context");
+ assertThat(context).isNotNull();
+ assertThat(context.get("message")).isNull();
+ assertThat(context.get("db")).isNull();
+ assertThat(context.get("http")).isNull();
 
  span.withType("template").withSubtype("jsf.lifecycle").withAction("render.view");
  spanJson = readJsonString(serializer.toJsonString(span));
@@ -268,6 +273,57 @@ void testSpanTypeSerialization() {
  assertThat(spanJson.get("type").isNull()).isTrue();
  }
 
+ @Test
+ void testSpanHttpContextSerialization() {
+ Span span = new Span(MockTracer.create());
+ span.getContext().getHttp()
+ .withMethod("GET")
+ .withStatusCode(523)
+ .withUrl("http://whatever.com/path");
+
+ JsonNode spanJson = readJsonString(serializer.toJsonString(span));
+ JsonNode context = spanJson.get("context");
+ JsonNode http = context.get("http");
+ assertThat(http).isNotNull();
+ assertThat(http.get("method").textValue()).isEqualTo("GET");
+ assertThat(http.get("url").textValue()).isEqualTo("http://whatever.com/path");
+ assertThat(http.get("status_code").intValue()).isEqualTo(523);
+ }
+
+ @Test
+ void testSpanMessageContextSerialization() {
+ Span span = new Span(MockTracer.create());
+ span.getContext().getMessage()
+ .withTopic("test-topic");
+
+ JsonNode spanJson = readJsonString(serializer.toJsonString(span));
+ JsonNode context = spanJson.get("context");
+ JsonNode message = context.get("message");
+ assertThat(message).isNotNull();
+ JsonNode queue = message.get("queue");
+ assertThat(queue).isNull();
+ JsonNode topic = message.get("topic");
+ assertThat(topic).isNotNull();
+ assertThat("test-topic").isEqualTo(topic.get("name").textValue());
+ }
+
+ @Test
+ void testSpanDbContextSerialization() {
+ Span span = new Span(MockTracer.create());
+ span.getContext().getDb()
+ .withAffectedRowsCount(5)
+ .withInstance("test-instance")
+ .withStatement("SELECT * FROM TABLE").withDbLink("db-link");
+
+ JsonNode spanJson = readJsonString(serializer.toJsonString(span));
+ JsonNode context = spanJson.get("context");
+ JsonNode db = context.get("db");
+ assertThat(db).isNotNull();
+ assertThat(db.get("rows_affected").longValue()).isEqualTo(5);
+ assertThat(db.get("instance").textValue()).isEqualTo("test-instance");
+ assertThat(db.get("statement").textValue()).isEqualTo("SELECT * FROM TABLE");
+ }
+
  @Test
  void testInlineReplacement() {
  StringBuilder sb = new StringBuilder("this.is.a.string");
@@ -368,7 +424,7 @@ void testConfiguredServiceNodeName() {
  }
 
  @Test
- void testTransactionContext() {
+ void testTransactionContextSerialization() {
 
  ElasticApmTracer tracer = MockTracer.create();
  Transaction transaction = new Transaction(tracer);
@@ -405,6 +461,8 @@ void testTransactionContext() {
  .addHeader("response_header", "value")
  .withStatusCode(418);
 
+ transaction.getContext().getMessage().withQueue("test_queue");
+
  String jsonString = serializer.toJsonString(transaction);
  JsonNode json = readJsonString(jsonString);
 
@@ -444,6 +502,13 @@ void testTransactionContext() {
  assertThat(jsonResponse.get("headers_sent").asBoolean()).isFalse();
  assertThat(jsonResponse.get("status_code").asInt()).isEqualTo(418);
 
+ JsonNode message = jsonContext.get("message");
+ assertThat(message).isNotNull();
+ JsonNode topic = message.get("topic");
+ assertThat(topic).isNull();
+ JsonNode queue = message.get("queue");
+ assertThat(queue).isNotNull();
+ assertThat("test_queue").isEqualTo(queue.get("name").textValue());
  }
 
  private JsonNode readJsonString(String jsonString) {

apm-agent-plugins/apm-es-restclient-plugin/apm-es-restclient-plugin-common/src/test/java/co/elastic/apm/agent/es/restclient/AbstractEsClientInstrumentationTest.java

@@ -26,8 +26,8 @@
 
 import co.elastic.apm.agent.AbstractInstrumentationTest;
 import co.elastic.apm.agent.impl.error.ErrorCapture;
-import co.elastic.apm.agent.impl.transaction.Db;
-import co.elastic.apm.agent.impl.transaction.Http;
+import co.elastic.apm.agent.impl.context.Db;
+import co.elastic.apm.agent.impl.context.Http;
 import co.elastic.apm.agent.impl.transaction.Span;
 import co.elastic.apm.agent.impl.transaction.TraceContext;
 import co.elastic.apm.agent.impl.transaction.Transaction;

apm-agent-plugins/apm-jdbc-plugin/src/test/java/co/elastic/apm/agent/jdbc/AbstractJdbcInstrumentationTest.java

@@ -25,7 +25,7 @@
 package co.elastic.apm.agent.jdbc;
 
 import co.elastic.apm.agent.AbstractInstrumentationTest;
-import co.elastic.apm.agent.impl.transaction.Db;
+import co.elastic.apm.agent.impl.context.Db;
 import co.elastic.apm.agent.impl.transaction.Span;
 import co.elastic.apm.agent.impl.transaction.TraceContext;
 import co.elastic.apm.agent.impl.transaction.Transaction;