x-pack/plugin/otel: introduce x-pack-otel plugin

.github/CODEOWNERS

@@ -27,8 +27,12 @@ libs/logstash-bridge @elastic/logstash
 x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/store/KibanaOwnedReservedRoleDescriptors.java @elastic/kibana-security
 
 # APM Data index templates, etc.
-x-pack/plugin/apm-data/src/main/resources @elastic/apm-server
-x-pack/plugin/apm-data/src/yamlRestTest/resources @elastic/apm-server
+x-pack/plugin/apm-data/src/main/resources @elastic/obs-ds-intake-services
+x-pack/plugin/apm-data/src/yamlRestTest/resources @elastic/obs-ds-intake-services
+
+# OTel
+x-pack/plugin/otel-data/src/main/resources @elastic/obs-ds-intake-services
+x-pack/plugin/otel-data/src/yamlRestTest/resources @elastic/obs-ds-intake-services
 
 # Delivery
 gradle @elastic/es-delivery

docs/changelog/111091.yaml

@@ -0,0 +1,5 @@
+pr: 111091
+summary: "X-pack/plugin/otel: introduce x-pack-otel plugin"
+area: Data streams
+type: feature
+issues: []

x-pack/plugin/apm-data/src/main/java/org/elasticsearch/xpack/apmdata/APMIndexTemplateRegistry.java

@@ -7,53 +7,24 @@
 
 package org.elasticsearch.xpack.apmdata;
 
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
 import org.elasticsearch.client.internal.Client;
-import org.elasticsearch.cluster.ClusterChangedEvent;
-import org.elasticsearch.cluster.metadata.ComponentTemplate;
-import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.xcontent.XContentHelper;
-import org.elasticsearch.core.Nullable;
 import org.elasticsearch.features.FeatureService;
-import org.elasticsearch.features.NodeFeature;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
-import org.elasticsearch.xcontent.XContentParserConfiguration;
-import org.elasticsearch.xcontent.yaml.YamlXContent;
 import org.elasticsearch.xpack.core.ClientHelper;
-import org.elasticsearch.xpack.core.template.IndexTemplateRegistry;
-import org.elasticsearch.xpack.core.template.IngestPipelineConfig;
+import org.elasticsearch.xpack.core.template.YamlTemplateRegistry;
 
-import java.io.IOException;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.stream.Collectors;
-
-import static org.elasticsearch.xpack.apmdata.ResourceUtils.APM_TEMPLATE_VERSION_VARIABLE;
-import static org.elasticsearch.xpack.apmdata.ResourceUtils.loadResource;
-import static org.elasticsearch.xpack.apmdata.ResourceUtils.loadVersionedResourceUTF8;
+import static org.elasticsearch.xpack.apmdata.APMPlugin.APM_DATA_REGISTRY_ENABLED;
 
 /**
  * Creates all index templates and ingest pipelines that are required for using Elastic APM.
  */
-public class APMIndexTemplateRegistry extends IndexTemplateRegistry {
-    private static final Logger logger = LogManager.getLogger(APMIndexTemplateRegistry.class);
-    // this node feature is a redefinition of {@link DataStreamFeatures#DATA_STREAM_LIFECYCLE} and it's meant to avoid adding a
-    // dependency to the data-streams module just for this
-    public static final NodeFeature DATA_STREAM_LIFECYCLE = new NodeFeature("data_stream.lifecycle");
-    private final int version;
+public class APMIndexTemplateRegistry extends YamlTemplateRegistry {
 
-    private final Map<String, ComponentTemplate> componentTemplates;
-    private final Map<String, ComposableIndexTemplate> composableIndexTemplates;
-    private final List<IngestPipelineConfig> ingestPipelines;
-    private final FeatureService featureService;
-    private volatile boolean enabled;
+    public static final String APM_TEMPLATE_VERSION_VARIABLE = "xpack.apmdata.template.version";
 
-    @SuppressWarnings("unchecked")
     public APMIndexTemplateRegistry(
         Settings nodeSettings,
         ClusterService clusterService,
@@ -62,133 +33,29 @@ public APMIndexTemplateRegistry(
         NamedXContentRegistry xContentRegistry,
         FeatureService featureService
     ) {
-        super(nodeSettings, clusterService, threadPool, client, xContentRegistry);
-
-        try {
-            final Map<String, Object> apmResources = XContentHelper.convertToMap(
-                YamlXContent.yamlXContent,
-                loadResource("/resources.yaml"),
-                false
-            );
-            version = (((Number) apmResources.get("version")).intValue());
-            final List<Object> componentTemplateNames = (List<Object>) apmResources.get("component-templates");
-            final List<Object> indexTemplateNames = (List<Object>) apmResources.get("index-templates");
-            final List<Object> ingestPipelineConfigs = (List<Object>) apmResources.get("ingest-pipelines");
-
-            componentTemplates = componentTemplateNames.stream()
-                .map(o -> (String) o)
-                .collect(Collectors.toMap(name -> name, name -> loadComponentTemplate(name, version)));
-            composableIndexTemplates = indexTemplateNames.stream()
-                .map(o -> (String) o)
-                .collect(Collectors.toMap(name -> name, name -> loadIndexTemplate(name, version)));
-            ingestPipelines = ingestPipelineConfigs.stream().map(o -> (Map<String, Map<String, Object>>) o).map(map -> {
-                Map.Entry<String, Map<String, Object>> pipelineConfig = map.entrySet().iterator().next();
-                return loadIngestPipeline(pipelineConfig.getKey(), version, (List<String>) pipelineConfig.getValue().get("dependencies"));
-            }).collect(Collectors.toList());
-            this.featureService = featureService;
-        } catch (IOException e) {
-            throw new RuntimeException(e);
-        }
-    }
-
-    public int getVersion() {
-        return version;
-    }
-
-    void setEnabled(boolean enabled) {
-        logger.info("APM index template registry is {}", enabled ? "enabled" : "disabled");
-        this.enabled = enabled;
-    }
-
-    public boolean isEnabled() {
-        return enabled;
-    }
-
-    public void close() {
-        clusterService.removeListener(this);
-    }
-
-    @Override
-    protected String getOrigin() {
-        return ClientHelper.APM_ORIGIN;
-    }
-
-    @Override
-    protected boolean isClusterReady(ClusterChangedEvent event) {
-        // Ensure current version of the components are installed only after versions that support data stream lifecycle
-        // due to the use of the feature in all the `@lifecycle` component templates
-        return featureService.clusterHasFeature(event.state(), DATA_STREAM_LIFECYCLE);
+        super(nodeSettings, clusterService, threadPool, client, xContentRegistry, featureService);
     }
 
     @Override
-    protected boolean requiresMasterNode() {
-        return true;
+    public String getName() {
+        return "apm";
     }
 
     @Override
-    protected Map<String, ComponentTemplate> getComponentTemplateConfigs() {
-        if (enabled) {
-            return componentTemplates;
-        } else {
-            return Map.of();
+    public void initialize() {
+        super.initialize();
+        if (isEnabled()) {
+            clusterService.getClusterSettings().addSettingsUpdateConsumer(APM_DATA_REGISTRY_ENABLED, this::setEnabled);
         }
     }
 
     @Override
-    protected Map<String, ComposableIndexTemplate> getComposableTemplateConfigs() {
-        if (enabled) {
-            return composableIndexTemplates;
-        } else {
-            return Map.of();
-        }
-    }
-
-    @Override
-    protected List<IngestPipelineConfig> getIngestPipelines() {
-        if (enabled) {
-            return ingestPipelines;
-        } else {
-            return Collections.emptyList();
-        }
-    }
-
-    private static ComponentTemplate loadComponentTemplate(String name, int version) {
-        try {
-            final byte[] content = loadVersionedResourceUTF8("/component-templates/" + name + ".yaml", version);
-            try (var parser = YamlXContent.yamlXContent.createParser(XContentParserConfiguration.EMPTY, content)) {
-                return ComponentTemplate.parse(parser);
-            }
-        } catch (Exception e) {
-            throw new RuntimeException("failed to load APM Ingest plugin's component template: " + name, e);
-        }
-    }
-
-    private static ComposableIndexTemplate loadIndexTemplate(String name, int version) {
-        try {
-            final byte[] content = loadVersionedResourceUTF8("/index-templates/" + name + ".yaml", version);
-            try (var parser = YamlXContent.yamlXContent.createParser(XContentParserConfiguration.EMPTY, content)) {
-                return ComposableIndexTemplate.parse(parser);
-            }
-        } catch (Exception e) {
-            throw new RuntimeException("failed to load APM Ingest plugin's index template: " + name, e);
-        }
-    }
-
-    private static IngestPipelineConfig loadIngestPipeline(String name, int version, @Nullable List<String> dependencies) {
-        if (dependencies == null) {
-            dependencies = Collections.emptyList();
-        }
-        return new YamlIngestPipelineConfig(
-            name,
-            "/ingest-pipelines/" + name + ".yaml",
-            version,
-            APM_TEMPLATE_VERSION_VARIABLE,
-            dependencies
-        );
+    protected String getVersionProperty() {
+        return APM_TEMPLATE_VERSION_VARIABLE;
     }
 
     @Override
-    protected boolean applyRolloverAfterTemplateV2Upgrade() {
-        return true;
+    protected String getOrigin() {
+        return ClientHelper.APM_ORIGIN;
     }
 }

x-pack/plugin/apm-data/src/main/java/org/elasticsearch/xpack/apmdata/APMPlugin.java

@@ -60,7 +60,6 @@ public Collection<?> createComponents(PluginServices services) {
         if (enabled) {
             APMIndexTemplateRegistry registryInstance = registry.get();
             registryInstance.setEnabled(APM_DATA_REGISTRY_ENABLED.get(settings));
-            clusterService.getClusterSettings().addSettingsUpdateConsumer(APM_DATA_REGISTRY_ENABLED, registryInstance::setEnabled);
             registryInstance.initialize();
         }
         return Collections.emptyList();

x-pack/plugin/apm-data/src/main/resources/resources.yaml

@@ -23,7 +23,6 @@ component-templates:
   - metrics-apm.service_summary@mappings
   - metrics-apm.service_transaction@mappings
   - metrics-apm.transaction@mappings
-  - traces@mappings
   - traces-apm@mappings
   - traces-apm.rum@mappings
 

x-pack/plugin/apm-data/src/test/java/org/elasticsearch/xpack/apmdata/APMIndexTemplateRegistryTests.java

@@ -75,7 +75,6 @@
 public class APMIndexTemplateRegistryTests extends ESTestCase {
     private APMIndexTemplateRegistry apmIndexTemplateRegistry;
     private StackTemplateRegistryAccessor stackTemplateRegistryAccessor;
-    private ClusterService clusterService;
     private ThreadPool threadPool;
     private VerifyingClient client;
 
@@ -89,7 +88,7 @@ public void createRegistryAndClient() {
 
         threadPool = new TestThreadPool(this.getClass().getName());
         client = new VerifyingClient(threadPool);
-        clusterService = ClusterServiceUtils.createClusterService(threadPool, clusterSettings);
+        ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool, clusterSettings);
         FeatureService featureService = new FeatureService(List.of(new DataStreamFeatures()));
         stackTemplateRegistryAccessor = new StackTemplateRegistryAccessor(
             new StackTemplateRegistry(Settings.EMPTY, clusterService, threadPool, client, NamedXContentRegistry.EMPTY, featureService)

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ClientHelper.java

@@ -194,6 +194,7 @@ private static String maybeRewriteSingleAuthenticationHeaderForVersion(
     public static final String CONNECTORS_ORIGIN = "connectors";
     public static final String INFERENCE_ORIGIN = "inference";
     public static final String APM_ORIGIN = "apm";
+    public static final String OTEL_ORIGIN = "otel";
 
     private ClientHelper() {}
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackSettings.java

@@ -93,6 +93,13 @@ public Iterator<Setting<?>> settings() {
     /** Setting for enabling or disabling APM Data. Defaults to true. */
     public static final Setting<Boolean> APM_DATA_ENABLED = Setting.boolSetting("xpack.apm_data.enabled", true, Setting.Property.NodeScope);
 
+    /** Setting for enabling or disabling OTel Data. Defaults to true. */
+    public static final Setting<Boolean> OTEL_DATA_ENABLED = Setting.boolSetting(
+        "xpack.otel_data.enabled",
+        true,
+        Setting.Property.NodeScope
+    );
+
     /** Setting for enabling or disabling enterprise search. Defaults to true. */
     public static final Setting<Boolean> ENTERPRISE_SEARCH_ENABLED = Setting.boolSetting(
         "xpack.ent_search.enabled",

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/template/ResourceUtils.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.template;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+
+public class ResourceUtils {
+    static byte[] loadVersionedResourceUTF8(Class<?> clazz, String name, int version, String versionProperty) {
+        return loadVersionedResourceUTF8(clazz, name, version, versionProperty, Map.of());
+    }
+
+    static byte[] loadVersionedResourceUTF8(
+        Class<?> clazz,
+        String name,
+        int version,
+        String versionProperty,
+        Map<String, String> variables
+    ) {
+        try {
+            String content = loadResource(clazz, name);
+            content = TemplateUtils.replaceVariables(content, String.valueOf(version), versionProperty, variables);
+            return content.getBytes(StandardCharsets.UTF_8);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public static String loadResource(Class<?> clazz, String name) throws IOException {
+        InputStream is = clazz.getResourceAsStream(name);
+        if (is == null) {
+            throw new IOException("Resource [" + name + "] not found in classpath.");
+        }
+        return new String(is.readAllBytes(), java.nio.charset.StandardCharsets.UTF_8);
+    }
+
+}