Merge branch 'main' into es-gpu

Author: mayya-sharipova

build-tools/src/main/java/org/elasticsearch/gradle/testclusters/ElasticsearchCluster.java

@@ -64,7 +64,7 @@
 public class ElasticsearchCluster implements TestClusterConfiguration, Named {
 
     private static final Logger LOGGER = Logging.getLogger(ElasticsearchNode.class);
-    private static final int CLUSTER_UP_TIMEOUT = 40;
+    private static final int CLUSTER_UP_TIMEOUT = 120;
     private static final TimeUnit CLUSTER_UP_TIMEOUT_UNIT = TimeUnit.SECONDS;
 
     private final AtomicBoolean configurationFrozen = new AtomicBoolean(false);

docs/changelog/135262.yaml

@@ -0,0 +1,5 @@
+pr: 135262
+summary: Add usage stats for `semantic_text` fields
+area: "Vector Search"
+type: enhancement
+issues: []

docs/changelog/135635.yaml

@@ -0,0 +1,5 @@
+pr: 135635
+summary: Add executor name attribute to cache miss metrics
+area: Search
+type: enhancement
+issues: []

muted-tests.yml

@@ -588,9 +588,6 @@ tests:
 - class: org.elasticsearch.xpack.esql.qa.single_node.EsqlSpecIT
   method: test {csv-spec:spatial_shapes.ConvertCartesianShapeFromStringParseError}
   issue: https://github.com/elastic/elasticsearch/issues/135455
-- class: org.elasticsearch.xpack.esql.qa.multi_node.GenerativeIT
-  method: test
-  issue: https://github.com/elastic/elasticsearch/issues/134407
 - class: org.elasticsearch.cluster.routing.allocation.decider.RestoreInProgressAllocationDeciderTests
   method: testCanAllocatePrimaryExistingInRestoreInProgress
   issue: https://github.com/elastic/elasticsearch/issues/135566
@@ -603,9 +600,6 @@ tests:
 - class: org.elasticsearch.xpack.esql.heap_attack.HeapAttackIT
   method: testAggTooManyMvLongs
   issue: https://github.com/elastic/elasticsearch/issues/135585
-- class: org.elasticsearch.xpack.esql.qa.single_node.GenerativeIT
-  method: test
-  issue: https://github.com/elastic/elasticsearch/issues/134407
 
 # Examples:
 #

server/src/main/resources/transport/definitions/referable/inference_telemetry_added_semantic_text_stats.csv

@@ -0,0 +1 @@
+9182000

server/src/main/resources/transport/upper_bounds/9.2.csv

@@ -1 +1 @@
-index_reshard_shardcount_small,9181000
+inference_telemetry_added_semantic_text_stats,9182000

x-pack/plugin/blob-cache/src/main/java/org/elasticsearch/blobcache/BlobCacheMetrics.java

@@ -9,6 +9,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.index.store.LuceneFilesExtensions;
 import org.elasticsearch.telemetry.TelemetryProvider;
 import org.elasticsearch.telemetry.metric.DoubleHistogram;
@@ -29,7 +30,9 @@ public class BlobCacheMetrics {
     public static final String CACHE_POPULATION_REASON_ATTRIBUTE_KEY = "reason";
     public static final String CACHE_POPULATION_SOURCE_ATTRIBUTE_KEY = "source";
     public static final String LUCENE_FILE_EXTENSION_ATTRIBUTE_KEY = "file_extension";
+    public static final String ES_EXECUTOR_ATTRIBUTE_KEY = "executor";
     public static final String NON_LUCENE_EXTENSION_TO_RECORD = "other";
+    public static final String NON_ES_EXECUTOR_TO_RECORD = "other";
     public static final String BLOB_CACHE_COUNT_OF_EVICTED_REGIONS_TOTAL = "es.blob_cache.count_of_evicted_regions.total";
     public static final String SEARCH_ORIGIN_REMOTE_STORAGE_DOWNLOAD_TOOK_TIME = "es.blob_cache.search_origin.download_took_time.total";
 
@@ -198,13 +201,16 @@ public void recordCachePopulationMetrics(
     ) {
         LuceneFilesExtensions luceneFilesExtensions = LuceneFilesExtensions.fromFile(fileName);
         String luceneFileExt = luceneFilesExtensions != null ? luceneFilesExtensions.getExtension() : NON_LUCENE_EXTENSION_TO_RECORD;
+        String executorName = EsExecutors.executorName(Thread.currentThread());
         Map<String, Object> metricAttributes = Map.of(
             CACHE_POPULATION_REASON_ATTRIBUTE_KEY,
             cachePopulationReason.name(),
             CACHE_POPULATION_SOURCE_ATTRIBUTE_KEY,
             cachePopulationSource.name(),
             LUCENE_FILE_EXTENSION_ATTRIBUTE_KEY,
-            luceneFileExt
+            luceneFileExt,
+            ES_EXECUTOR_ATTRIBUTE_KEY,
+            executorName != null ? executorName : NON_ES_EXECUTOR_TO_RECORD
         );
         assert bytesCopied > 0 : "We shouldn't be recording zero-sized copies";
         cachePopulationBytes.incrementBy(bytesCopied, metricAttributes);

x-pack/plugin/blob-cache/src/main/java/org/elasticsearch/blobcache/shared/SharedBlobCacheService.java

@@ -29,6 +29,7 @@
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.RelativeByteSizeValue;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.util.concurrent.ThrottledTaskRunner;
 import org.elasticsearch.core.AbstractRefCounted;
 import org.elasticsearch.core.Assertions;
@@ -69,7 +70,9 @@
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
+import static org.elasticsearch.blobcache.BlobCacheMetrics.ES_EXECUTOR_ATTRIBUTE_KEY;
 import static org.elasticsearch.blobcache.BlobCacheMetrics.LUCENE_FILE_EXTENSION_ATTRIBUTE_KEY;
+import static org.elasticsearch.blobcache.BlobCacheMetrics.NON_ES_EXECUTOR_TO_RECORD;
 import static org.elasticsearch.blobcache.BlobCacheMetrics.NON_LUCENE_EXTENSION_TO_RECORD;
 
 /**
@@ -1284,6 +1287,7 @@ public void fillCacheRange(
                     ActionListener<Void> completionListener
                 ) throws IOException {
                     String blobFileExtension = getFileExtension(resourceDescription);
+                    String executorName = EsExecutors.executorName(Thread.currentThread());
                     writer.fillCacheRange(
                         channel,
                         channelPos,
@@ -1295,7 +1299,15 @@ public void fillCacheRange(
                             var elapsedTime = TimeUnit.NANOSECONDS.toMillis(relativeTimeInNanosSupplier.getAsLong() - startTime);
                             blobCacheMetrics.getCacheMissLoadTimes().record(elapsedTime);
                             blobCacheMetrics.getCacheMissCounter()
-                                .incrementBy(1L, Map.of(LUCENE_FILE_EXTENSION_ATTRIBUTE_KEY, blobFileExtension));
+                                .incrementBy(
+                                    1L,
+                                    Map.of(
+                                        LUCENE_FILE_EXTENSION_ATTRIBUTE_KEY,
+                                        blobFileExtension,
+                                        ES_EXECUTOR_ATTRIBUTE_KEY,
+                                        executorName != null ? executorName : NON_ES_EXECUTOR_TO_RECORD
+                                    )
+                                );
                             return null;
                         })
                     );

x-pack/plugin/blob-cache/src/test/java/org/elasticsearch/blobcache/BlobCacheMetricsTests.java

@@ -19,6 +19,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.stream.IntStream;
 
+import static org.elasticsearch.blobcache.BlobCacheMetrics.NON_ES_EXECUTOR_TO_RECORD;
 import static org.hamcrest.Matchers.is;
 
 public class BlobCacheMetricsTests extends ESTestCase {
@@ -46,27 +47,28 @@ public void testRecordCachePopulationMetricsRecordsThroughput() {
             cachePopulationReason,
             cachePopulationSource
         );
+        String threadName = NON_ES_EXECUTOR_TO_RECORD;
 
         // throughput histogram
         Measurement throughputMeasurement = recordingMeterRegistry.getRecorder()
             .getMeasurements(InstrumentType.DOUBLE_HISTOGRAM, "es.blob_cache.population.throughput.histogram")
             .get(0);
         assertEquals(throughputMeasurement.getDouble(), (double) mebiBytesSent / secondsTaken, 0.0);
-        assertExpectedAttributesPresent(throughputMeasurement, cachePopulationReason, cachePopulationSource, fileExtension);
+        assertExpectedAttributesPresent(throughputMeasurement, cachePopulationReason, cachePopulationSource, fileExtension, threadName);
 
         // bytes counter
         Measurement totalBytesMeasurement = recordingMeterRegistry.getRecorder()
             .getMeasurements(InstrumentType.LONG_COUNTER, "es.blob_cache.population.bytes.total")
             .get(0);
         assertEquals(totalBytesMeasurement.getLong(), ByteSizeValue.ofMb(mebiBytesSent).getBytes());
-        assertExpectedAttributesPresent(totalBytesMeasurement, cachePopulationReason, cachePopulationSource, fileExtension);
+        assertExpectedAttributesPresent(totalBytesMeasurement, cachePopulationReason, cachePopulationSource, fileExtension, threadName);
 
         // time counter
         Measurement totalTimeMeasurement = recordingMeterRegistry.getRecorder()
             .getMeasurements(InstrumentType.LONG_COUNTER, "es.blob_cache.population.time.total")
             .get(0);
         assertEquals(totalTimeMeasurement.getLong(), TimeUnit.SECONDS.toMillis(secondsTaken));
-        assertExpectedAttributesPresent(totalTimeMeasurement, cachePopulationReason, cachePopulationSource, fileExtension);
+        assertExpectedAttributesPresent(totalTimeMeasurement, cachePopulationReason, cachePopulationSource, fileExtension, threadName);
 
         // let us check for 0, avoid div by 0.
         checkReadsAndMisses(0, 0, 1);
@@ -107,10 +109,12 @@ private static void assertExpectedAttributesPresent(
         Measurement measurement,
         BlobCacheMetrics.CachePopulationReason cachePopulationReason,
         CachePopulationSource cachePopulationSource,
-        String fileExtension
+        String fileExtension,
+        String threadName
     ) {
         assertThat(measurement.attributes().get(BlobCacheMetrics.CACHE_POPULATION_REASON_ATTRIBUTE_KEY), is(cachePopulationReason.name()));
         assertThat(measurement.attributes().get(BlobCacheMetrics.CACHE_POPULATION_SOURCE_ATTRIBUTE_KEY), is(cachePopulationSource.name()));
         assertThat(measurement.attributes().get(BlobCacheMetrics.LUCENE_FILE_EXTENSION_ATTRIBUTE_KEY), is(fileExtension));
+        assertThat(measurement.attributes().get(BlobCacheMetrics.ES_EXECUTOR_ATTRIBUTE_KEY), is(threadName));
     }
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/inference/usage/ModelStats.java

@@ -7,9 +7,12 @@
 
 package org.elasticsearch.xpack.core.inference.usage;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.core.Nullable;
+import org.elasticsearch.features.NodeFeature;
 import org.elasticsearch.inference.TaskType;
 import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
@@ -19,28 +22,34 @@
 
 public class ModelStats implements ToXContentObject, Writeable {
 
+    public static final NodeFeature SEMANTIC_TEXT_USAGE = new NodeFeature("inference.semantic_text_usage");
+
+    static final TransportVersion INFERENCE_TELEMETRY_ADDED_SEMANTIC_TEXT_STATS = TransportVersion.fromName(
+        "inference_telemetry_added_semantic_text_stats"
+    );
+
     private final String service;
     private final TaskType taskType;
     private long count;
+    @Nullable
+    private final SemanticTextStats semanticTextStats;
 
-    public ModelStats(String service, TaskType taskType) {
-        this(service, taskType, 0L);
-    }
-
-    public ModelStats(String service, TaskType taskType, long count) {
+    public ModelStats(String service, TaskType taskType, long count, @Nullable SemanticTextStats semanticTextStats) {
         this.service = service;
         this.taskType = taskType;
         this.count = count;
-    }
-
-    public ModelStats(ModelStats stats) {
-        this(stats.service, stats.taskType, stats.count);
+        this.semanticTextStats = semanticTextStats;
     }
 
     public ModelStats(StreamInput in) throws IOException {
         this.service = in.readString();
         this.taskType = in.readEnum(TaskType.class);
         this.count = in.readLong();
+        if (in.getTransportVersion().supports(INFERENCE_TELEMETRY_ADDED_SEMANTIC_TEXT_STATS)) {
+            this.semanticTextStats = in.readOptional(SemanticTextStats::new);
+        } else {
+            this.semanticTextStats = null;
+        }
     }
 
     public void add() {
@@ -59,6 +68,11 @@ public long count() {
         return count;
     }
 
+    @Nullable
+    public SemanticTextStats semanticTextStats() {
+        return semanticTextStats;
+    }
+
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
@@ -71,25 +85,34 @@ public void addXContentFragment(XContentBuilder builder, Params params) throws I
         builder.field("service", service);
         builder.field("task_type", taskType.name());
         builder.field("count", count);
+        if (semanticTextStats != null) {
+            builder.field("semantic_text", semanticTextStats);
+        }
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeString(service);
         out.writeEnum(taskType);
         out.writeLong(count);
+        if (out.getTransportVersion().supports(INFERENCE_TELEMETRY_ADDED_SEMANTIC_TEXT_STATS)) {
+            out.writeOptionalWriteable(semanticTextStats);
+        }
     }
 
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;
         if (o == null || getClass() != o.getClass()) return false;
         ModelStats that = (ModelStats) o;
-        return count == that.count && Objects.equals(service, that.service) && taskType == that.taskType;
+        return count == that.count
+            && Objects.equals(service, that.service)
+            && taskType == that.taskType
+            && Objects.equals(semanticTextStats, that.semanticTextStats);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(service, taskType, count);
+        return Objects.hash(service, taskType, count, semanticTextStats);
     }
 }