Skip to content

Commit dc8c35d

Browse files
astefanastefan
authored
Merge branch 'main' into inlinestats_pickup3
2 parents eddde2d + 4ff1aad commit dc8c35d

File tree

19 files changed

+1263
-75
lines changed

19 files changed

+1263
-75
lines changed

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

Lines changed: 39 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -61,11 +61,14 @@
6161
import java.io.UncheckedIOException;
6262
import java.net.URL;
6363
import java.nio.charset.StandardCharsets;
64+
import java.nio.file.FileVisitResult;
6465
import java.nio.file.Files;
6566
import java.nio.file.NoSuchFileException;
6667
import java.nio.file.Path;
68+
import java.nio.file.SimpleFileVisitor;
6769
import java.nio.file.StandardCopyOption;
6870
import java.nio.file.StandardOpenOption;
71+
import java.nio.file.attribute.BasicFileAttributes;
6972
import java.time.Instant;
7073
import java.util.ArrayList;
7174
import java.util.Arrays;
@@ -1295,40 +1298,47 @@ private void syncWithCopy(Path sourceRoot, Path destinationRoot) {
12951298

12961299
private void sync(Path sourceRoot, Path destinationRoot, BiConsumer<Path, Path> syncMethod) {
12971300
assert Files.exists(destinationRoot) == false;
1298-
try (Stream<Path> stream = Files.walk(sourceRoot)) {
1299-
stream.forEach(source -> {
1300-
Path relativeDestination = sourceRoot.relativize(source);
1301-
if (relativeDestination.getNameCount() <= 1) {
1302-
return;
1303-
}
1304-
// Throw away the first name as the archives have everything in a single top level folder we are not interested in
1305-
relativeDestination = relativeDestination.subpath(1, relativeDestination.getNameCount());
1306-
1307-
Path destination = destinationRoot.resolve(relativeDestination);
1308-
if (Files.isDirectory(source)) {
1309-
try {
1310-
Files.createDirectories(destination);
1311-
} catch (IOException e) {
1312-
throw new UncheckedIOException("Can't create directory " + destination.getParent(), e);
1301+
try {
1302+
Files.walkFileTree(sourceRoot, new SimpleFileVisitor<>() {
1303+
@Override
1304+
public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
1305+
Path relativeDestination = sourceRoot.relativize(dir);
1306+
if (relativeDestination.getNameCount() <= 1) {
1307+
return FileVisitResult.CONTINUE;
13131308
}
1314-
} else {
1315-
try {
1316-
Files.createDirectories(destination.getParent());
1317-
} catch (IOException e) {
1318-
throw new UncheckedIOException("Can't create directory " + destination.getParent(), e);
1309+
// Throw away the first name as the archives have everything in a single top level folder we are not interested in
1310+
relativeDestination = relativeDestination.subpath(1, relativeDestination.getNameCount());
1311+
Path destination = destinationRoot.resolve(relativeDestination);
1312+
Files.createDirectories(destination);
1313+
return FileVisitResult.CONTINUE;
1314+
}
1315+
1316+
@Override
1317+
public FileVisitResult visitFile(Path source, BasicFileAttributes attrs) throws IOException {
1318+
Path relativeDestination = sourceRoot.relativize(source);
1319+
if (relativeDestination.getNameCount() <= 1) {
1320+
return FileVisitResult.CONTINUE;
13191321
}
1322+
// Throw away the first name as the archives have everything in a single top level folder we are not interested in
1323+
relativeDestination = relativeDestination.subpath(1, relativeDestination.getNameCount());
1324+
Path destination = destinationRoot.resolve(relativeDestination);
1325+
Files.createDirectories(destination.getParent());
13201326
syncMethod.accept(destination, source);
1327+
return FileVisitResult.CONTINUE;
13211328
}
1322-
});
1323-
} catch (UncheckedIOException e) {
1324-
if (e.getCause() instanceof NoSuchFileException cause) {
1325-
// Ignore these files that are sometimes left behind by the JVM
1326-
if (cause.getFile() == null || cause.getFile().contains(".attach_pid") == false) {
1327-
throw new UncheckedIOException(cause);
1329+
1330+
@Override
1331+
public FileVisitResult visitFileFailed(Path file, IOException exc) throws IOException {
1332+
if (exc instanceof NoSuchFileException noFileException) {
1333+
// Ignore these files that are sometimes left behind by the JVM
1334+
if (noFileException.getFile() != null && noFileException.getFile().contains(".attach_pid")) {
1335+
LOGGER.info("Ignoring file left behind by JVM: {}", noFileException.getFile());
1336+
return FileVisitResult.CONTINUE;
1337+
}
1338+
}
1339+
throw exc;
13281340
}
1329-
} else {
1330-
throw e;
1331-
}
1341+
});
13321342
} catch (IOException e) {
13331343
throw new UncheckedIOException("Can't walk source " + sourceRoot, e);
13341344
}

‎docs/changelog/122218.yaml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
pr: 122218
2+
summary: Integrate with `DeepSeek` API
3+
area: Machine Learning
4+
type: enhancement
5+
issues: []

‎server/src/main/java/org/elasticsearch/TransportVersions.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -147,6 +147,7 @@ static TransportVersion def(int id) {
147147
public static final TransportVersion JINA_AI_EMBEDDING_TYPE_SUPPORT_ADDED_BACKPORT_8_19 = def(8_841_0_06);
148148
public static final TransportVersion RETRY_ILM_ASYNC_ACTION_REQUIRE_ERROR_8_19 = def(8_841_0_07);
149149
public static final TransportVersion INFERENCE_CONTEXT_8_X = def(8_841_0_08);
150+
public static final TransportVersion ML_INFERENCE_DEEPSEEK_8_19 = def(8_841_0_09);
150151
public static final TransportVersion INITIAL_ELASTICSEARCH_9_0 = def(9_000_0_00);
151152
public static final TransportVersion REMOVE_SNAPSHOT_FAILURES_90 = def(9_000_0_01);
152153
public static final TransportVersion TRANSPORT_STATS_HANDLING_TIME_REQUIRED_90 = def(9_000_0_02);
@@ -183,6 +184,7 @@ static TransportVersion def(int id) {
183184
public static final TransportVersion ESQL_SERIALIZE_BLOCK_TYPE_CODE = def(9_026_0_00);
184185
public static final TransportVersion ESQL_THREAD_NAME_IN_DRIVER_PROFILE = def(9_027_0_00);
185186
public static final TransportVersion INFERENCE_CONTEXT = def(9_028_0_00);
187+
public static final TransportVersion ML_INFERENCE_DEEPSEEK = def(9_029_00_0);
186188

187189
/*
188190
* STOP! READ THIS FIRST! No, really,

‎test/test-clusters/build.gradle‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,10 @@ dependencies {
99
implementation "com.fasterxml.jackson.core:jackson-annotations:${versions.jackson}"
1010
implementation "com.fasterxml.jackson.core:jackson-databind:${versions.jackson}"
1111
implementation "org.elasticsearch.gradle:reaper"
12+
13+
testImplementation "junit:junit:${versions.junit}"
14+
testImplementation "org.hamcrest:hamcrest:${versions.hamcrest}"
15+
testImplementation "org.apache.logging.log4j:log4j-core:${versions.log4j}"
1216
}
1317

1418
tasks.named("processResources").configure {

‎test/test-clusters/src/main/java/org/elasticsearch/test/cluster/util/IOUtils.java‎

Lines changed: 31 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -15,9 +15,12 @@
1515
import java.io.File;
1616
import java.io.IOException;
1717
import java.io.UncheckedIOException;
18+
import java.nio.file.FileVisitResult;
1819
import java.nio.file.Files;
1920
import java.nio.file.NoSuchFileException;
2021
import java.nio.file.Path;
22+
import java.nio.file.SimpleFileVisitor;
23+
import java.nio.file.attribute.BasicFileAttributes;
2124
import java.util.Comparator;
2225
import java.util.function.BiConsumer;
2326
import java.util.stream.Stream;
@@ -118,35 +121,37 @@ public static void syncWithCopy(Path sourceRoot, Path destinationRoot) {
118121

119122
private static void sync(Path sourceRoot, Path destinationRoot, BiConsumer<Path, Path> syncMethod) {
120123
assert Files.exists(destinationRoot) == false;
121-
try (Stream<Path> stream = Files.walk(sourceRoot)) {
122-
stream.forEach(source -> {
123-
Path relativeDestination = sourceRoot.relativize(source);
124-
125-
Path destination = destinationRoot.resolve(relativeDestination);
126-
if (Files.isDirectory(source)) {
127-
try {
128-
Files.createDirectories(destination);
129-
} catch (IOException e) {
130-
throw new UncheckedIOException("Can't create directory " + destination.getParent(), e);
131-
}
132-
} else {
133-
try {
134-
Files.createDirectories(destination.getParent());
135-
} catch (IOException e) {
136-
throw new UncheckedIOException("Can't create directory " + destination.getParent(), e);
137-
}
124+
try {
125+
Files.walkFileTree(sourceRoot, new SimpleFileVisitor<>() {
126+
@Override
127+
public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
128+
Path relativeDestination = sourceRoot.relativize(dir);
129+
Path destination = destinationRoot.resolve(relativeDestination);
130+
Files.createDirectories(destination);
131+
return FileVisitResult.CONTINUE;
132+
}
133+
134+
@Override
135+
public FileVisitResult visitFile(Path source, BasicFileAttributes attrs) throws IOException {
136+
Path relativeDestination = sourceRoot.relativize(source);
137+
Path destination = destinationRoot.resolve(relativeDestination);
138+
Files.createDirectories(destination.getParent());
138139
syncMethod.accept(destination, source);
140+
return FileVisitResult.CONTINUE;
139141
}
140-
});
141-
} catch (UncheckedIOException e) {
142-
if (e.getCause() instanceof NoSuchFileException cause) {
143-
// Ignore these files that are sometimes left behind by the JVM
144-
if (cause.getFile() == null || cause.getFile().contains(".attach_pid") == false) {
145-
throw new UncheckedIOException(cause);
142+
143+
@Override
144+
public FileVisitResult visitFileFailed(Path file, IOException exc) throws IOException {
145+
if (exc instanceof NoSuchFileException noFileException) {
146+
// Ignore these files that are sometimes left behind by the JVM
147+
if (noFileException.getFile() != null && noFileException.getFile().contains(".attach_pid")) {
148+
LOGGER.info("Ignoring file left behind by JVM: {}", noFileException.getFile());
149+
return FileVisitResult.CONTINUE;
150+
}
151+
}
152+
throw exc;
146153
}
147-
} else {
148-
throw e;
149-
}
154+
});
150155
} catch (IOException e) {
151156
throw new UncheckedIOException("Can't walk source " + sourceRoot, e);
152157
}

‎test/test-clusters/src/test/java/org/elasticsearch/test/cluster/util/IOUtilsTests.java‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
/*
2+
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
3+
* or more contributor license agreements. Licensed under the "Elastic License
4+
* 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
5+
* Public License v 1"; you may not use this file except in compliance with, at
6+
* your election, the "Elastic License 2.0", the "GNU Affero General Public
7+
* License v3.0 only", or the "Server Side Public License, v 1".
8+
*/
9+
10+
package org.elasticsearch.test.cluster.util;
11+
12+
import org.junit.Test;
13+
14+
import java.io.IOException;
15+
import java.io.UncheckedIOException;
16+
import java.nio.file.Files;
17+
import java.nio.file.Path;
18+
19+
import static org.hamcrest.CoreMatchers.containsString;
20+
import static org.hamcrest.MatcherAssert.assertThat;
21+
import static org.hamcrest.core.Is.is;
22+
import static org.hamcrest.core.Is.isA;
23+
import static org.junit.Assert.assertThrows;
24+
25+
public class IOUtilsTests {
26+
27+
@Test
28+
public void testSyncWithLinks() throws IOException {
29+
// given
30+
Path sourceDir = Files.createTempDirectory("sourceDir");
31+
Files.createFile(sourceDir.resolve("file1.txt"));
32+
Files.createFile(sourceDir.resolve("file2.txt"));
33+
Files.createDirectory(sourceDir.resolve("nestedDir"));
34+
Files.createFile(sourceDir.resolve("nestedDir").resolve("file3.txt"));
35+
36+
Path baseDestinationDir = Files.createTempDirectory("baseDestinationDir");
37+
Path destinationDir = baseDestinationDir.resolve("destinationDir");
38+
39+
// when
40+
IOUtils.syncWithLinks(sourceDir, destinationDir);
41+
42+
// then
43+
assertFileExists(destinationDir.resolve("file1.txt"));
44+
assertFileExists(destinationDir.resolve("file2.txt"));
45+
assertFileExists(destinationDir.resolve("nestedDir").resolve("file3.txt"));
46+
}
47+
48+
private void assertFileExists(Path path) throws IOException {
49+
assertThat("File " + path + " doesn't exist", Files.exists(path), is(true));
50+
assertThat("File " + path + " is not a regular file", Files.isRegularFile(path), is(true));
51+
assertThat("File " + path + " is not readable", Files.isReadable(path), is(true));
52+
if (OS.current() != OS.WINDOWS) {
53+
assertThat("Expected 2 hard links", Files.getAttribute(path, "unix:nlink"), is(2));
54+
}
55+
}
56+
57+
@Test
58+
public void testSyncWithLinksThrowExceptionWhenDestinationIsNotWritable() throws IOException {
59+
// given
60+
Path sourceDir = Files.createTempDirectory("sourceDir");
61+
Files.createFile(sourceDir.resolve("file1.txt"));
62+
63+
Path baseDestinationDir = Files.createTempDirectory("baseDestinationDir");
64+
Path destinationDir = baseDestinationDir.resolve("destinationDir");
65+
66+
baseDestinationDir.toFile().setWritable(false);
67+
68+
// when
69+
UncheckedIOException ex = assertThrows(UncheckedIOException.class, () -> IOUtils.syncWithLinks(sourceDir, destinationDir));
70+
71+
// then
72+
assertThat(ex.getCause(), isA(IOException.class));
73+
assertThat(ex.getCause().getMessage(), containsString("destinationDir"));
74+
}
75+
}

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

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99

1010
import org.apache.logging.log4j.LogManager;
1111
import org.apache.logging.log4j.Logger;
12+
import org.elasticsearch.index.store.LuceneFilesExtensions;
1213
import org.elasticsearch.telemetry.TelemetryProvider;
1314
import org.elasticsearch.telemetry.metric.DoubleHistogram;
1415
import org.elasticsearch.telemetry.metric.LongCounter;
@@ -24,8 +25,8 @@ public class BlobCacheMetrics {
2425
private static final double BYTES_PER_NANOSECONDS_TO_MEBIBYTES_PER_SECOND = 1e9D / (1 << 20);
2526
public static final String CACHE_POPULATION_REASON_ATTRIBUTE_KEY = "reason";
2627
public static final String CACHE_POPULATION_SOURCE_ATTRIBUTE_KEY = "source";
27-
public static final String SHARD_ID_ATTRIBUTE_KEY = "shard_id";
28-
public static final String INDEX_ATTRIBUTE_KEY = "index_name";
28+
public static final String LUCENE_FILE_EXTENSION_ATTRIBUTE_KEY = "file_extension";
29+
public static final String NON_LUCENE_EXTENSION_TO_RECORD = "other";
2930

3031
private final LongCounter cacheMissCounter;
3132
private final LongCounter evictedCountNonZeroFrequency;
@@ -113,22 +114,28 @@ public LongHistogram getCacheMissLoadTimes() {
113114
/**
114115
* Record the various cache population metrics after a chunk is copied to the cache
115116
*
117+
* @param blobName The file that was requested and triggered the cache population.
116118
* @param bytesCopied The number of bytes copied
117119
* @param copyTimeNanos The time taken to copy the bytes in nanoseconds
118120
* @param cachePopulationReason The reason for the cache being populated
119121
* @param cachePopulationSource The source from which the data is being loaded
120122
*/
121123
public void recordCachePopulationMetrics(
124+
String blobName,
122125
int bytesCopied,
123126
long copyTimeNanos,
124127
CachePopulationReason cachePopulationReason,
125128
CachePopulationSource cachePopulationSource
126129
) {
130+
LuceneFilesExtensions luceneFilesExtensions = LuceneFilesExtensions.fromFile(blobName);
131+
String blobFileExtension = luceneFilesExtensions != null ? luceneFilesExtensions.getExtension() : NON_LUCENE_EXTENSION_TO_RECORD;
127132
Map<String, Object> metricAttributes = Map.of(
128133
CACHE_POPULATION_REASON_ATTRIBUTE_KEY,
129134
cachePopulationReason.name(),
130135
CACHE_POPULATION_SOURCE_ATTRIBUTE_KEY,
131-
cachePopulationSource.name()
136+
cachePopulationSource.name(),
137+
LUCENE_FILE_EXTENSION_ATTRIBUTE_KEY,
138+
blobFileExtension
132139
);
133140
assert bytesCopied > 0 : "We shouldn't be recording zero-sized copies";
134141
cachePopulationBytes.incrementBy(bytesCopied, metricAttributes);

0 commit comments

Comments
 (0)