Skip to content

Add xpack.apm.tracing.names.include setting for filtering #80871

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Nov 22, 2021
Merged

Add xpack.apm.tracing.names.include setting for filtering #80871

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 100 additions & 0 deletions ...lugin/apm-integration/src/internalClusterTest/java/org/elasticsearch/xpack/apm/ApmIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
import io.opentelemetry.api.common.AttributeKey;
import io.opentelemetry.sdk.trace.data.SpanData;

import org.elasticsearch.action.admin.cluster.node.stats.NodesStatsRequest;
import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.index.IndexRequestBuilder;
Expand Down Expand Up @@ -40,9 +41,11 @@
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static java.util.stream.Collectors.toList;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasSize;
Expand Down Expand Up @@ -227,4 +230,101 @@ public void testDoesNotRecordSpansWhenDisabled() {
.actionGet();
}
}

public void testFilterByNameGivenSingleCompleteMatch() {

client().admin()
.cluster()
.updateSettings(
new ClusterUpdateSettingsRequest().persistentSettings(
Settings.builder().put(APMTracer.APM_TRACING_NAMES_INCLUDE_SETTING.getKey(), "cluster:monitor/tasks/lists").build()
)
)
.actionGet();

APMTracer.CAPTURING_SPAN_EXPORTER.clear();// removing start related events

try {
client().admin().cluster().prepareListTasks().get();

var parentTasks = APMTracer.CAPTURING_SPAN_EXPORTER.findSpanByName("cluster:monitor/tasks/lists").collect(toList());
assertThat(parentTasks, hasSize(1));

var childrenTasks = APMTracer.CAPTURING_SPAN_EXPORTER.findSpanByName("cluster:monitor/tasks/lists[n]").collect(toList());
assertThat(childrenTasks, empty());
} finally {
client().admin()
.cluster()
.updateSettings(
new ClusterUpdateSettingsRequest().persistentSettings(
Settings.builder().put(APMTracer.APM_TRACING_NAMES_INCLUDE_SETTING.getKey(), (String) null).build()
)
)
.actionGet();
}
}

public void testFilterByNameGivenSinglePattern() {

client().admin()
.cluster()
.updateSettings(
new ClusterUpdateSettingsRequest().persistentSettings(
Settings.builder().put(APMTracer.APM_TRACING_NAMES_INCLUDE_SETTING.getKey(), "*/tasks/lists*").build()
)
)
.actionGet();

APMTracer.CAPTURING_SPAN_EXPORTER.clear();// removing start related events

try {
client().admin().cluster().prepareListTasks().get();

var parentTasks = APMTracer.CAPTURING_SPAN_EXPORTER.findSpanByName("cluster:monitor/tasks/lists").collect(toList());
assertThat(parentTasks, hasSize(1));

var childrenTasks = APMTracer.CAPTURING_SPAN_EXPORTER.findSpanByName("cluster:monitor/tasks/lists[n]").collect(toList());
assertThat(childrenTasks, hasSize(internalCluster().size()));
} finally {
client().admin()
.cluster()
.updateSettings(
new ClusterUpdateSettingsRequest().persistentSettings(
Settings.builder().put(APMTracer.APM_TRACING_NAMES_INCLUDE_SETTING.getKey(), (String) null).build()
)
)
.actionGet();
}
}

public void testFilterByNameGivenTwoPatterns() {

client().admin()
.cluster()
.updateSettings(
new ClusterUpdateSettingsRequest().persistentSettings(
Settings.builder().put(APMTracer.APM_TRACING_NAMES_INCLUDE_SETTING.getKey(), "*/tasks/lists,*/nodes/stats").build()
)
)
.actionGet();

APMTracer.CAPTURING_SPAN_EXPORTER.clear();// removing start related events

try {
client().admin().cluster().prepareListTasks().get();
client().admin().cluster().nodesStats(new NodesStatsRequest()).actionGet();

var spans = APMTracer.CAPTURING_SPAN_EXPORTER.getCapturedSpans().stream().map(SpanData::getName).collect(Collectors.toSet());
assertThat(spans, contains("cluster:monitor/nodes/stats", "cluster:monitor/tasks/lists"));
} finally {
client().admin()
.cluster()
.updateSettings(
new ClusterUpdateSettingsRequest().persistentSettings(
Settings.builder().put(APMTracer.APM_TRACING_NAMES_INCLUDE_SETTING.getKey(), (String) null).build()
)
)
.actionGet();
}
}
}
7 changes: 6 additions & 1 deletion x-pack/plugin/apm-integration/src/main/java/org/elasticsearch/xpack/apm/APM.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,12 @@ public Collection<Object> createComponents(

@Override
public List<Setting<?>> getSettings() {
return List.of(APMTracer.APM_ENABLED_SETTING, APMTracer.APM_ENDPOINT_SETTING, APMTracer.APM_TOKEN_SETTING);
return List.of(
APMTracer.APM_ENABLED_SETTING,
APMTracer.APM_ENDPOINT_SETTING,
APMTracer.APM_TOKEN_SETTING,
APMTracer.APM_TRACING_NAMES_INCLUDE_SETTING
);
}

public List<TransportInterceptor> getTransportInterceptors(NamedWriteableRegistry namedWriteableRegistry, ThreadContext threadContext) {
Expand Down
29 changes: 29 additions & 0 deletions x-pack/plugin/apm-integration/src/main/java/org/elasticsearch/xpack/apm/APMTracer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.component.AbstractLifecycleComponent;
import org.elasticsearch.common.regex.Regex;
import org.elasticsearch.common.settings.SecureSetting;
import org.elasticsearch.common.settings.SecureString;
import org.elasticsearch.common.settings.Setting;
Expand All @@ -45,13 +46,15 @@
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Queue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;
Expand All @@ -66,6 +69,13 @@ public class APMTracer extends AbstractLifecycleComponent implements org.elastic
static final Setting<Boolean> APM_ENABLED_SETTING = Setting.boolSetting("xpack.apm.tracing.enabled", false, Dynamic, NodeScope);
static final Setting<SecureString> APM_ENDPOINT_SETTING = SecureSetting.secureString("xpack.apm.endpoint", null);
static final Setting<SecureString> APM_TOKEN_SETTING = SecureSetting.secureString("xpack.apm.token", null);
static final Setting<List<String>> APM_TRACING_NAMES_INCLUDE_SETTING = Setting.listSetting(
"xpack.apm.tracing.names.include",
Collections.emptyList(),
Function.identity(),
Dynamic,
NodeScope
);

private final Semaphore shutdownPermits = new Semaphore(Integer.MAX_VALUE);
private final Map<String, Span> spans = ConcurrentCollections.newConcurrentMap();
Expand All @@ -77,6 +87,8 @@ public class APMTracer extends AbstractLifecycleComponent implements org.elastic
private volatile boolean enabled;
private volatile APMServices services;

private List<String> includeNames;

/** This class is required to make all open telemetry services visible at once */
private static class APMServices {
private final SdkTracerProvider provider;
Expand All @@ -96,7 +108,9 @@ public APMTracer(Settings settings, ThreadPool threadPool, ClusterService cluste
this.endpoint = APM_ENDPOINT_SETTING.get(settings);
this.token = APM_TOKEN_SETTING.get(settings);
this.enabled = APM_ENABLED_SETTING.get(settings);
this.includeNames = APM_TRACING_NAMES_INCLUDE_SETTING.get(settings);
clusterService.getClusterSettings().addSettingsUpdateConsumer(APM_ENABLED_SETTING, this::setEnabled);
clusterService.getClusterSettings().addSettingsUpdateConsumer(APM_TRACING_NAMES_INCLUDE_SETTING, this::setIncludeNames);
}

public boolean isEnabled() {
Expand All @@ -112,6 +126,10 @@ private void setEnabled(boolean enabled) {
}
}

private void setIncludeNames(List<String> includeNames) {
this.includeNames = includeNames;
}

@Override
protected void doStart() {
if (enabled) {
Expand Down Expand Up @@ -189,6 +207,11 @@ public void onTraceStarted(Traceable traceable) {
if (services == null) {
return;
}

if (isSpanNameIncluded(traceable.getSpanName()) == false) {
return;
}

spans.computeIfAbsent(traceable.getSpanId(), spanId -> {
// services might be in shutdown sate by this point, but this is handled by the open telemetry internally
final SpanBuilder spanBuilder = services.tracer.spanBuilder(traceable.getSpanName());
Expand Down Expand Up @@ -222,6 +245,12 @@ public void onTraceStarted(Traceable traceable) {
});
}

private boolean isSpanNameIncluded(String name) {
// Alternatively we could use automata here but it is much more complex
// and it needs wrapping like done for use in the security plugin.
return includeNames.isEmpty() || Regex.simpleMatch(includeNames, name);
}

private Context getParentSpanContext(OpenTelemetry openTelemetry) {
// If we already have a non-root span context that should be the parent
if (Context.current() != Context.root()) {
Expand Down