From 6680dc622d59815beccaed695a347fbedbed4bec Mon Sep 17 00:00:00 2001 From: minwoox Date: Wed, 1 Jul 2026 10:12:35 +0900 Subject: [PATCH 1/3] Add YAML/JSON backward compatibility for xDS resource files Motivation: - xDS resources are currently stored as .json files, but are planned to be migrated to .yaml format. - To support safe rollout and rollback, the system must handle both .json and .yaml files coexisting in the same repository without errors. Modifications: - XdsResourceWatchingService: accept EntryType.YAML alongside JSON during the initial repository scan; handle UPSERT_YAML change type in the diff watcher; normalize YAML content to JSON before invoking handleXdsResource() so all implementations continue to use JSON_MESSAGE_MARSHALLER without format-specific logic. - XdsResourceManager: add alternativeFileName() helper that swaps between .json and .yaml extensions; change updateOrDelete() to search for both the requested filename and its alternative so that update and delete operations locate the file regardless of which format it was stored in; pass the resolved filename back to the task via Function so the correct file is mutated; make push() emit Change.ofYamlUpsert when the resolved filename ends with .yaml. Result: - .yaml and .json xDS resource files can coexist in the same group repository; the control plane, CRUD API, and endpoint read API all handle both formats transparently. --- .../server/command/ContentTransformer.java | 3 +- .../git/TransformingChangesApplier.java | 17 +- .../v1/XdsEndpointUpdateScheduler.java | 25 +- .../internal/CentralDogmaXdsResources.java | 9 +- .../xds/internal/ControlPlaneService.java | 10 +- .../xds/internal/XdsEndpointReadService.java | 24 +- .../xds/internal/XdsResourceManager.java | 35 ++- .../internal/XdsResourceWatchingService.java | 12 +- .../XdsKubernetesEndpointFetchingService.java | 13 +- .../endpoint/v1/XdsEndpointServiceTest.java | 69 +++++ .../endpoint/v1/XdsRegisterEndpointTest.java | 57 ++++ .../internal/XdsEndpointReadServiceTest.java | 265 +++++++++++++++++ .../XdsResourceWatchingServiceTest.java | 144 ++++++++- .../internal/XdsYamlCompatibilityTest.java | 281 ++++++++++++++++++ 14 files changed, 915 insertions(+), 49 deletions(-) create mode 100644 xds/src/test/java/com/linecorp/centraldogma/xds/internal/XdsEndpointReadServiceTest.java create mode 100644 xds/src/test/java/com/linecorp/centraldogma/xds/internal/XdsYamlCompatibilityTest.java diff --git a/server/src/main/java/com/linecorp/centraldogma/server/command/ContentTransformer.java b/server/src/main/java/com/linecorp/centraldogma/server/command/ContentTransformer.java index 3e8d75d09..8fd45cbe5 100644 --- a/server/src/main/java/com/linecorp/centraldogma/server/command/ContentTransformer.java +++ b/server/src/main/java/com/linecorp/centraldogma/server/command/ContentTransformer.java @@ -40,7 +40,8 @@ public class ContentTransformer { */ public ContentTransformer(String path, EntryType entryType, BiFunction transformer) { this.path = requireNonNull(path, "path"); - checkArgument(entryType == EntryType.JSON, "entryType: %s (expected: %s)", entryType, EntryType.JSON); + checkArgument(entryType == EntryType.JSON || entryType == EntryType.YAML, + "entryType: %s (expected: JSON or YAML)", entryType); this.entryType = requireNonNull(entryType, "entryType"); this.transformer = requireNonNull(transformer, "transformer"); } diff --git a/server/src/main/java/com/linecorp/centraldogma/server/internal/storage/repository/git/TransformingChangesApplier.java b/server/src/main/java/com/linecorp/centraldogma/server/internal/storage/repository/git/TransformingChangesApplier.java index 20c208852..943ef6956 100644 --- a/server/src/main/java/com/linecorp/centraldogma/server/internal/storage/repository/git/TransformingChangesApplier.java +++ b/server/src/main/java/com/linecorp/centraldogma/server/internal/storage/repository/git/TransformingChangesApplier.java @@ -35,6 +35,7 @@ import com.linecorp.centraldogma.common.EntryType; import com.linecorp.centraldogma.common.Revision; import com.linecorp.centraldogma.internal.Jackson; +import com.linecorp.centraldogma.internal.Yaml; import com.linecorp.centraldogma.server.command.ContentTransformer; final class TransformingChangesApplier extends AbstractChangesApplier { @@ -42,8 +43,9 @@ final class TransformingChangesApplier extends AbstractChangesApplier { private final ContentTransformer transformer; TransformingChangesApplier(ContentTransformer transformer) { - checkArgument(transformer.entryType() == EntryType.JSON, - "transformer: %s (expected: JSON type)", transformer); + checkArgument(transformer.entryType() == EntryType.JSON || + transformer.entryType() == EntryType.YAML, + "transformer: %s (expected: JSON or YAML type)", transformer); //noinspection unchecked this.transformer = (ContentTransformer) transformer; } @@ -55,13 +57,20 @@ int doApply(Revision headRevision, DirCache dirCache, final DirCacheEntry oldEntry = dirCache.getEntry(changePath); final byte[] oldContent = oldEntry != null ? reader.open(oldEntry.getObjectId()).getBytes() : null; - final JsonNode oldJsonNode = oldContent != null ? Jackson.readTree(oldContent) + final boolean isYaml = transformer.entryType() == EntryType.YAML; + final JsonNode oldJsonNode = oldContent != null ? (isYaml ? Yaml.readTree(oldContent) + : Jackson.readTree(oldContent)) : JsonNodeFactory.instance.nullNode(); try { final JsonNode newJsonNode = transformer.transformer().apply(headRevision, oldJsonNode.deepCopy()); requireNonNull(newJsonNode, "transformer.transformer().apply() returned null"); if (!Objects.equals(newJsonNode, oldJsonNode)) { - applyPathEdit(dirCache, new InsertJson(changePath, inserter, newJsonNode)); + if (isYaml) { + applyPathEdit(dirCache, new InsertText(changePath, inserter, + Yaml.writeValueAsString(newJsonNode))); + } else { + applyPathEdit(dirCache, new InsertJson(changePath, inserter, newJsonNode)); + } return 1; } } catch (CentralDogmaException e) { diff --git a/xds/src/main/java/com/linecorp/centraldogma/xds/endpoint/v1/XdsEndpointUpdateScheduler.java b/xds/src/main/java/com/linecorp/centraldogma/xds/endpoint/v1/XdsEndpointUpdateScheduler.java index 79752c27b..38067fea8 100644 --- a/xds/src/main/java/com/linecorp/centraldogma/xds/endpoint/v1/XdsEndpointUpdateScheduler.java +++ b/xds/src/main/java/com/linecorp/centraldogma/xds/endpoint/v1/XdsEndpointUpdateScheduler.java @@ -81,6 +81,16 @@ int batchUpdateTaskSize() { return batchUpdateTasks.size(); } + private static String alternativeFileName(String fileName) { + if (fileName.endsWith(".json")) { + return fileName.substring(0, fileName.length() - 5) + ".yaml"; + } + if (fileName.endsWith(".yaml")) { + return fileName.substring(0, fileName.length() - 5) + ".json"; + } + return fileName; + } + void schedule(String group, String endpointName, String fileName, LocalityLbEndpoint localityLbEndpoint, StreamObserver streamObserver, boolean register) { final EndpointIdentifier identifier = EndpointIdentifier.of(localityLbEndpoint); @@ -170,11 +180,10 @@ private void flush() { } } - final ContentTransformer transformer = new ContentTransformer<>( - fileName, EntryType.JSON, new BatchUpdateTransformer(toRegister, toDeregister)); - final Repository repository = xdsResourceManager.xdsProject().repos().get(group); - repository.find(Revision.HEAD, fileName, FIND_ONE_WITHOUT_CONTENT).handle((entries, cause) -> { + final String altFileName = alternativeFileName(fileName); + repository.find(Revision.HEAD, fileName + ',' + altFileName, FIND_ONE_WITHOUT_CONTENT) + .handle((entries, cause) -> { if (cause != null) { copied.forEach(pendingUpdate -> pendingUpdate.streamObserver.onError(cause)); return null; @@ -186,6 +195,11 @@ private void flush() { copied.forEach(pendingUpdate -> pendingUpdate.streamObserver.onError(runtimeException)); return null; } + final String resolvedFileName = entries.keySet().iterator().next(); + final EntryType entryType = + resolvedFileName.endsWith(".yaml") ? EntryType.YAML : EntryType.JSON; + final ContentTransformer transformer = new ContentTransformer<>( + resolvedFileName, entryType, new BatchUpdateTransformer(toRegister, toDeregister)); final String commitMessage = "Batch update for " + endpointName + " in group " + group + ": " + toRegister.size() + " register, " + toDeregister.size() + " deregister"; @@ -304,8 +318,7 @@ private static ClusterLoadAssignment.Builder toClusterLoadAssignmentBuilder(Json final Builder clusterLoadAssignmentBuilder = ClusterLoadAssignment.newBuilder(); try { - JSON_MESSAGE_MARSHALLER.mergeValue(Jackson.writeValueAsString(oldJsonNode), - clusterLoadAssignmentBuilder); + JSON_MESSAGE_MARSHALLER.mergeValue(oldJsonNode.traverse(), clusterLoadAssignmentBuilder); } catch (Throwable t) { // Should never reach here. throw new Error(); diff --git a/xds/src/main/java/com/linecorp/centraldogma/xds/internal/CentralDogmaXdsResources.java b/xds/src/main/java/com/linecorp/centraldogma/xds/internal/CentralDogmaXdsResources.java index 8d1b340ec..88d872d03 100644 --- a/xds/src/main/java/com/linecorp/centraldogma/xds/internal/CentralDogmaXdsResources.java +++ b/xds/src/main/java/com/linecorp/centraldogma/xds/internal/CentralDogmaXdsResources.java @@ -103,7 +103,8 @@ void removeCluster(String groupName, String path) { } private static String getResourceName(String groupName, String path) { - return "groups/" + groupName + path.substring(0, path.length() - 5); // Remove .json + // Remove .json or .yaml (both 5 chars) + return "groups/" + groupName + path.substring(0, path.length() - 5); } void removeEndpoint(String groupName, String path) { @@ -112,11 +113,11 @@ void removeEndpoint(String groupName, String path) { if (groupEndpoints == null) { return; } - // e.g. /endpoints/foo-cluster.json file with group foo -> groups/foo/clusters/foo-cluster - // e.g. /k8s/endpoints/foo-cluster.json file with group foo -> groups/foo/k8s/clusters/foo-cluster + // e.g. /endpoints/foo-cluster.json/.yaml file with group foo -> groups/foo/clusters/foo-cluster + // e.g. /k8s/endpoints/foo-cluster.json/.yaml file with group foo -> groups/foo/k8s/clusters/foo-cluster final String clusterName = "groups/" + groupName + - ENDPOINTS_PATTERN.matcher(path.substring(0, path.length() - 5) /* remove .json */) + ENDPOINTS_PATTERN.matcher(path.substring(0, path.length() - 5) /* remove .json or .yaml */) .replaceFirst("/clusters/"); endpointUpdated |= groupEndpoints.remove(clusterName) != null; } diff --git a/xds/src/main/java/com/linecorp/centraldogma/xds/internal/ControlPlaneService.java b/xds/src/main/java/com/linecorp/centraldogma/xds/internal/ControlPlaneService.java index 376e13f1b..45f053fee 100644 --- a/xds/src/main/java/com/linecorp/centraldogma/xds/internal/ControlPlaneService.java +++ b/xds/src/main/java/com/linecorp/centraldogma/xds/internal/ControlPlaneService.java @@ -338,23 +338,23 @@ protected String pathPattern() { } @Override - protected void handleXdsResource(String path, String contentAsText, String groupName) + protected void handleXdsResource(String path, JsonNode content, String groupName) throws IOException { if (path.startsWith(CLUSTERS_DIRECTORY)) { final Cluster.Builder builder = Cluster.newBuilder(); - JSON_MESSAGE_MARSHALLER.mergeValue(contentAsText, builder); + JSON_MESSAGE_MARSHALLER.mergeValue(content.traverse(), builder); centralDogmaXdsResources.setCluster(groupName, builder.build()); } else if (path.startsWith(ENDPOINTS_DIRECTORY) || path.startsWith(K8S_ENDPOINTS_DIRECTORY)) { final ClusterLoadAssignment.Builder builder = ClusterLoadAssignment.newBuilder(); - JSON_MESSAGE_MARSHALLER.mergeValue(contentAsText, builder); + JSON_MESSAGE_MARSHALLER.mergeValue(content.traverse(), builder); centralDogmaXdsResources.setEndpoint(groupName, builder.build()); } else if (path.startsWith(LISTENERS_DIRECTORY)) { final Listener.Builder builder = Listener.newBuilder(); - JSON_MESSAGE_MARSHALLER.mergeValue(contentAsText, builder); + JSON_MESSAGE_MARSHALLER.mergeValue(content.traverse(), builder); centralDogmaXdsResources.setListener(groupName, builder.build()); } else if (path.startsWith(ROUTES_DIRECTORY)) { final RouteConfiguration.Builder builder = RouteConfiguration.newBuilder(); - JSON_MESSAGE_MARSHALLER.mergeValue(contentAsText, builder); + JSON_MESSAGE_MARSHALLER.mergeValue(content.traverse(), builder); centralDogmaXdsResources.setRoute(groupName, builder.build()); } else { // ignore diff --git a/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsEndpointReadService.java b/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsEndpointReadService.java index 4d4756f30..11916bddc 100644 --- a/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsEndpointReadService.java +++ b/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsEndpointReadService.java @@ -56,12 +56,12 @@ public CompletableFuture listEndpoints(@Param String group) { .thenApply(entries -> { final ArrayNode array = JsonNodeFactory.instance.arrayNode(); for (Entry entry : entries.values()) { - if (entry.type() != EntryType.JSON) { + if (entry.type() != EntryType.JSON && entry.type() != EntryType.YAML) { continue; } final ObjectNode node = array.addObject(); node.put("path", entry.path()); - node.put("type", "JSON"); + node.put("type", entry.type().name()); node.put("revision", entry.revision().major()); } return array; @@ -73,7 +73,7 @@ public CompletableFuture listEndpoints(@Param String group) { */ @Get("/xds/groups/{group}/endpoints/{*id}") public CompletableFuture getEndpoint(@Param String group, @Param String id) { - return read(group, ENDPOINTS_DIRECTORY + id + ".json"); + return readByBase(group, ENDPOINTS_DIRECTORY + id); } /** @@ -82,17 +82,27 @@ public CompletableFuture getEndpoint(@Param String group, @Param Strin */ @Get("/xds/groups/{group}/k8s/endpoints/{*id}") public CompletableFuture getK8sEndpoint(@Param String group, @Param String id) { - return read(group, K8S_ENDPOINTS_DIRECTORY + id + ".json"); + return readByBase(group, K8S_ENDPOINTS_DIRECTORY + id); } - private CompletableFuture read(String group, String path) { - return xdsProject.repos().get(group).get(Revision.HEAD, path).thenApply(XdsEndpointReadService::toNode); + private CompletableFuture readByBase(String group, String pathBase) { + final Repository repository = xdsProject.repos().get(group); + return repository.find(Revision.HEAD, pathBase + ".json," + pathBase + ".yaml") + .thenCompose(entries -> { + if (entries.isEmpty()) { + // Delegate to get() so the caller receives a proper EntryNotFoundException. + return repository.get(Revision.HEAD, pathBase + ".json") + .thenApply(XdsEndpointReadService::toNode); + } + return CompletableFuture.completedFuture( + toNode(entries.values().iterator().next())); + }); } private static JsonNode toNode(Entry entry) { final ObjectNode node = JsonNodeFactory.instance.objectNode(); node.put("path", entry.path()); - node.put("type", "JSON"); + node.put("type", entry.type().name()); node.put("revision", entry.revision().major()); node.set("content", (JsonNode) entry.content()); return node; diff --git a/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceManager.java b/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceManager.java index 03fcc1046..f3713507c 100644 --- a/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceManager.java +++ b/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceManager.java @@ -23,6 +23,7 @@ import java.io.IOException; import java.lang.reflect.Method; import java.lang.reflect.Modifier; +import java.util.function.Function; import java.util.regex.Pattern; import org.curioswitch.common.protobuf.json.MessageMarshaller; @@ -185,6 +186,8 @@ public void push( final JsonNode jsonNode = Jackson.readTree(jsonText); if (create) { change = Change.ofJsonPatch(fileName, null, jsonNode); + } else if (fileName.endsWith(".yaml")) { + change = Change.ofYamlUpsert(fileName, jsonNode); } else { change = Change.ofJsonUpsert(fileName, jsonNode); } @@ -236,8 +239,8 @@ public void update(StreamObserver responseObserver, Strin public void update(StreamObserver responseObserver, String group, String resourceName, String fileName, String summary, T resource, Author author) { - updateOrDelete(responseObserver, group, resourceName, fileName, - () -> push(responseObserver, group, resourceName, fileName, + updateOrDelete(responseObserver, group, resourceName, fileName, resolvedFileName -> + () -> push(responseObserver, group, resourceName, resolvedFileName, summary, resource, author, false)); } @@ -248,10 +251,10 @@ public void delete(StreamObserver responseObserver, String group, public void delete(StreamObserver responseObserver, String group, String resourceName, String fileName, String summary, Author author) { - final Runnable deleteTask = () -> + updateOrDelete(responseObserver, group, resourceName, fileName, resolvedFileName -> () -> commandExecutor.execute(Command.push(author, XDS_CENTRAL_DOGMA_PROJECT, group, Revision.HEAD, summary, "", Markup.PLAINTEXT, - ImmutableList.of(Change.ofRemoval(fileName)))) + ImmutableList.of(Change.ofRemoval(resolvedFileName)))) .handle((unused, cause) -> { if (cause != null) { responseObserver.onError( @@ -261,14 +264,17 @@ public void delete(StreamObserver responseObserver, String group, responseObserver.onNext(Empty.getDefaultInstance()); responseObserver.onCompleted(); return null; - }); - updateOrDelete(responseObserver, group, resourceName, fileName, deleteTask); + })); } public void updateOrDelete(StreamObserver responseObserver, String group, String resourceName, - String fileName, Runnable task) { + String fileName, Function taskProvider) { final Repository repository = xdsProject.repos().get(group); - repository.find(Revision.HEAD, fileName, FIND_ONE_WITHOUT_CONTENT).handle((entries, cause) -> { + // Search for both the requested filename and its alternative extension (.json ↔ .yaml) + // to support files that may have been written in either format. + final String altFileName = alternativeFileName(fileName); + repository.find(Revision.HEAD, fileName + ',' + altFileName, FIND_ONE_WITHOUT_CONTENT) + .handle((entries, cause) -> { if (cause != null) { responseObserver.onError(cause); return null; @@ -279,8 +285,19 @@ public void updateOrDelete(StreamObserver responseObserver, String group, Str .asRuntimeException()); return null; } - task.run(); + final String resolvedFileName = entries.keySet().iterator().next(); + taskProvider.apply(resolvedFileName).run(); return null; }); } + + private static String alternativeFileName(String fileName) { + if (fileName.endsWith(".json")) { + return fileName.substring(0, fileName.length() - 5) + ".yaml"; + } + if (fileName.endsWith(".yaml")) { + return fileName.substring(0, fileName.length() - 5) + ".json"; + } + return fileName; + } } diff --git a/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceWatchingService.java b/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceWatchingService.java index b0fb04160..dedfa3e93 100644 --- a/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceWatchingService.java +++ b/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceWatchingService.java @@ -26,6 +26,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.fasterxml.jackson.databind.JsonNode; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableList.Builder; import com.google.common.collect.Sets; @@ -70,7 +71,7 @@ protected Project xdsProject() { protected abstract String pathPattern(); - protected abstract void handleXdsResource(String path, String contentAsText, String groupName) + protected abstract void handleXdsResource(String path, JsonNode content, String groupName) throws IOException; protected abstract void onGroupRemoved(String groupName); @@ -110,13 +111,13 @@ protected void init() { " at revision: " + normalizedRevision, cause); } for (Entry entry : entries.values()) { - if (entry.type() != EntryType.JSON || !entry.hasContent()) { + if ((entry.type() != EntryType.JSON && entry.type() != EntryType.YAML) || + !entry.hasContent()) { continue; } final String path = entry.path(); - final String contentAsText = entry.contentAsText(); try { - handleXdsResource(path, contentAsText, groupName); + handleXdsResource(path, (JsonNode) entry.content(), groupName); } catch (Throwable t) { logger.warn("Unexpected exception while building an xDS resource from {}.", groupName + path, t); @@ -222,8 +223,9 @@ private void handleDiff(String groupName, Revision newRevision, final String path = change.path(); switch (change.type()) { case UPSERT_JSON: + case UPSERT_YAML: try { - handleXdsResource(path, change.contentAsText(), groupName); + handleXdsResource(path, (JsonNode) change.content(), groupName); } catch (Throwable t) { logger.warn("Unexpected exception while handling an xDS resource from {}.", groupName + path, t); diff --git a/xds/src/main/java/com/linecorp/centraldogma/xds/k8s/v1/XdsKubernetesEndpointFetchingService.java b/xds/src/main/java/com/linecorp/centraldogma/xds/k8s/v1/XdsKubernetesEndpointFetchingService.java index 2bd7ac7c6..9ef3d504e 100644 --- a/xds/src/main/java/com/linecorp/centraldogma/xds/k8s/v1/XdsKubernetesEndpointFetchingService.java +++ b/xds/src/main/java/com/linecorp/centraldogma/xds/k8s/v1/XdsKubernetesEndpointFetchingService.java @@ -44,7 +44,6 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.JsonNode; -import com.google.protobuf.InvalidProtocolBufferException; import com.spotify.futures.CompletableFutures; import com.linecorp.armeria.client.kubernetes.endpoints.KubernetesEndpointGroup; @@ -120,15 +119,15 @@ protected String pathPattern() { } @Override - protected void handleXdsResource(String path, String contentAsText, String groupName) - throws InvalidProtocolBufferException { + protected void handleXdsResource(String path, JsonNode content, String groupName) + throws IOException { final KubernetesEndpointAggregator.Builder aggregatorBuilder = KubernetesEndpointAggregator.newBuilder(); try { - JSON_MESSAGE_MARSHALLER.mergeValue(contentAsText, aggregatorBuilder); + JSON_MESSAGE_MARSHALLER.mergeValue(content.traverse(), aggregatorBuilder); } catch (IOException e) { logger.warn("Failed to parse a KubernetesEndpointAggregator at {}{}. content: {}", - groupName, path, contentAsText, e); + groupName, path, content, e); return; } @@ -179,8 +178,8 @@ protected void onGroupRemoved(String groupName) { protected void onFileRemoved(String groupName, String path) { final Map updaters = kubernetesEndpointsUpdaters.get(groupName); // e.g. groups/foo/k8s/endpointAggregators/foo-cluster - final String aggregatorName = - "groups/" + groupName + path.substring(0, path.length() - 5); // Remove .json + // Remove .json or .yaml (both 5 chars) + final String aggregatorName = "groups/" + groupName + path.substring(0, path.length() - 5); if (updaters != null) { final KubernetesEndpointsUpdater updater = updaters.get(aggregatorName); if (updater != null) { diff --git a/xds/src/test/java/com/linecorp/centraldogma/xds/endpoint/v1/XdsEndpointServiceTest.java b/xds/src/test/java/com/linecorp/centraldogma/xds/endpoint/v1/XdsEndpointServiceTest.java index d4c13f0ba..41d885fb0 100644 --- a/xds/src/test/java/com/linecorp/centraldogma/xds/endpoint/v1/XdsEndpointServiceTest.java +++ b/xds/src/test/java/com/linecorp/centraldogma/xds/endpoint/v1/XdsEndpointServiceTest.java @@ -15,6 +15,8 @@ */ package com.linecorp.centraldogma.xds.endpoint.v1; +import static com.linecorp.centraldogma.xds.internal.ControlPlanePlugin.XDS_CENTRAL_DOGMA_PROJECT; +import static com.linecorp.centraldogma.xds.internal.ControlPlaneService.ENDPOINTS_DIRECTORY; import static com.linecorp.centraldogma.xds.internal.XdsResourceManager.JSON_MESSAGE_MARSHALLER; import static com.linecorp.centraldogma.xds.internal.XdsTestUtil.createEndpoint; import static com.linecorp.centraldogma.xds.internal.XdsTestUtil.createGroup; @@ -46,6 +48,10 @@ import com.linecorp.armeria.common.HttpStatus; import com.linecorp.armeria.common.MediaType; import com.linecorp.armeria.common.RequestHeaders; +import com.linecorp.centraldogma.common.Change; +import com.linecorp.centraldogma.common.Revision; +import com.linecorp.centraldogma.server.storage.repository.FindOptions; +import com.linecorp.centraldogma.server.storage.repository.Repository; import com.linecorp.centraldogma.testing.junit.CentralDogmaExtension; import com.linecorp.centraldogma.xds.endpoint.v1.XdsEndpointServiceGrpc.XdsEndpointServiceBlockingStub; @@ -170,6 +176,69 @@ void updateEndpointViaHttp() throws Exception { checkEndpointsViaDiscoveryRequest(dogma.httpClient().uri(), actualEndpoint2, clusterName); } + @Test + void updateYamlEndpointViaHttp() throws Exception { + // Push an endpoint as YAML directly (simulating a JSON→YAML migration). + final String clusterName = "groups/foo/clusters/yaml-endpoint/1"; + final String endpointName = "groups/foo/endpoints/yaml-endpoint/1"; + final ClusterLoadAssignment initial = loadAssignment(clusterName, "127.0.0.1", 8080); + dogma.client().forRepo(XDS_CENTRAL_DOGMA_PROJECT, "foo") + .commit("Add YAML endpoint", + Change.ofYamlUpsert(ENDPOINTS_DIRECTORY + "yaml-endpoint/1.yaml", + JSON_MESSAGE_MARSHALLER.writeValueAsString(initial))) + .push().join(); + checkEndpointsViaDiscoveryRequest(dogma.httpClient().uri(), initial, clusterName); + + // Update via the HTTP API — updateOrDelete must locate the .yaml file, not create a new .json. + final ClusterLoadAssignment updated = + initial.toBuilder().setClusterName(clusterName) + .addEndpoints(LocalityLbEndpoints.newBuilder() + .addLbEndpoints(endpoint("127.0.0.1", 8081))) + .build(); + final AggregatedHttpResponse response = updateEndpoint("yaml-endpoint/1", updated); + assertOk(response); + + // The .yaml file must have been updated in-place; no new .json file should exist. + final Repository repo = + dogma.projectManager().get(XDS_CENTRAL_DOGMA_PROJECT).repos().get("foo"); + assertThat(repo.find(Revision.HEAD, ENDPOINTS_DIRECTORY + "yaml-endpoint/1.yaml", + FindOptions.FIND_ONE_WITHOUT_CONTENT).join()).isNotEmpty(); + assertThat(repo.find(Revision.HEAD, ENDPOINTS_DIRECTORY + "yaml-endpoint/1.json", + FindOptions.FIND_ONE_WITHOUT_CONTENT).join()).isEmpty(); + + final ClusterLoadAssignment.Builder endpointBuilder = ClusterLoadAssignment.newBuilder(); + JSON_MESSAGE_MARSHALLER.mergeValue(response.contentUtf8(), endpointBuilder); + checkEndpointsViaDiscoveryRequest(dogma.httpClient().uri(), endpointBuilder.build(), clusterName); + } + + @Test + void deleteYamlEndpointViaHttp() throws Exception { + // Push an endpoint as YAML directly. + final String clusterName = "groups/foo/clusters/yaml-endpoint/2"; + final String endpointName = "groups/foo/endpoints/yaml-endpoint/2"; + final ClusterLoadAssignment initial = loadAssignment(clusterName, "127.0.0.1", 8080); + dogma.client().forRepo(XDS_CENTRAL_DOGMA_PROJECT, "foo") + .commit("Add YAML endpoint", + Change.ofYamlUpsert(ENDPOINTS_DIRECTORY + "yaml-endpoint/2.yaml", + JSON_MESSAGE_MARSHALLER.writeValueAsString(initial))) + .push().join(); + checkEndpointsViaDiscoveryRequest(dogma.httpClient().uri(), initial, clusterName); + + // Delete via the HTTP API — updateOrDelete must locate and remove the .yaml file. + final AggregatedHttpResponse response = deleteEndpoint(endpointName); + assertOk(response); + assertThat(response.contentUtf8()).isEqualTo("{}"); + + // The .yaml file must be gone. + final Repository repo = + dogma.projectManager().get(XDS_CENTRAL_DOGMA_PROJECT).repos().get("foo"); + assertThat(repo.find(Revision.HEAD, ENDPOINTS_DIRECTORY + "yaml-endpoint/2.yaml", + FindOptions.FIND_ONE_WITHOUT_CONTENT).join()).isEmpty(); + + // Control plane must no longer serve the deleted endpoint. + checkEndpointsViaDiscoveryRequest(dogma.httpClient().uri(), null, clusterName); + } + private static AggregatedHttpResponse updateEndpoint( String endpointId, ClusterLoadAssignment endpoint) throws IOException { final RequestHeaders headers = RequestHeaders.builder(HttpMethod.PATCH, diff --git a/xds/src/test/java/com/linecorp/centraldogma/xds/endpoint/v1/XdsRegisterEndpointTest.java b/xds/src/test/java/com/linecorp/centraldogma/xds/endpoint/v1/XdsRegisterEndpointTest.java index cbf7c1365..ec2f4ece7 100644 --- a/xds/src/test/java/com/linecorp/centraldogma/xds/endpoint/v1/XdsRegisterEndpointTest.java +++ b/xds/src/test/java/com/linecorp/centraldogma/xds/endpoint/v1/XdsRegisterEndpointTest.java @@ -18,6 +18,7 @@ import static com.linecorp.centraldogma.xds.endpoint.v1.XdsEndpointServiceTest.assertOk; import static com.linecorp.centraldogma.xds.endpoint.v1.XdsEndpointServiceTest.checkEndpointsViaDiscoveryRequest; import static com.linecorp.centraldogma.xds.internal.ControlPlanePlugin.XDS_CENTRAL_DOGMA_PROJECT; +import static com.linecorp.centraldogma.xds.internal.ControlPlaneService.ENDPOINTS_DIRECTORY; import static com.linecorp.centraldogma.xds.internal.XdsResourceManager.JSON_MESSAGE_MARSHALLER; import static com.linecorp.centraldogma.xds.internal.XdsTestUtil.createEndpoint; import static com.linecorp.centraldogma.xds.internal.XdsTestUtil.createGroup; @@ -39,6 +40,7 @@ import com.linecorp.armeria.common.HttpStatus; import com.linecorp.armeria.common.MediaType; import com.linecorp.armeria.common.RequestHeaders; +import com.linecorp.centraldogma.common.Change; import com.linecorp.centraldogma.common.Revision; import com.linecorp.centraldogma.server.storage.repository.Repository; import com.linecorp.centraldogma.testing.junit.CentralDogmaExtension; @@ -278,4 +280,59 @@ void invalidRegister() throws IOException { // endpoint name is not found. assertThat(response.status()).isSameAs(HttpStatus.NOT_FOUND); } + + @Test + void registerAndDeregisterOnYamlEndpoint() throws Exception { + // Push the endpoint file directly as YAML (simulating a migration from JSON to YAML). + final String clusterName = "groups/foo/clusters/yaml-register-ep"; + final String endpointName = "groups/foo/endpoints/yaml-register-ep"; + final Locality locality = Locality.newBuilder().setRegion("r1").setZone("z1").build(); + final ClusterLoadAssignment initial = loadAssignment(clusterName, locality, + endpoint("127.0.0.1", 9100)); + dogma.client().forRepo(XDS_CENTRAL_DOGMA_PROJECT, "foo") + .commit("Add YAML endpoint", + Change.ofYamlUpsert(ENDPOINTS_DIRECTORY + "yaml-register-ep.yaml", + JSON_MESSAGE_MARSHALLER.writeValueAsString(initial))) + .push().join(); + + checkEndpointsViaDiscoveryRequest(dogma.httpClient().uri(), initial, clusterName); + + // Register a new lb endpoint into the YAML-backed endpoint file. + final LocalityLbEndpoint toRegister = + LocalityLbEndpoint.newBuilder().setLocality(locality) + .setLbEndpoint(endpoint("127.0.0.1", 9101)) + .build(); + final AggregatedHttpResponse registerResponse = + registerOrDeregister(endpointName, toRegister, true); + assertOk(registerResponse); + + final ClusterLoadAssignment expected = + initial.toBuilder() + .addEndpoints(LocalityLbEndpoints.newBuilder() + .setLocality(locality) + .addLbEndpoints(endpoint("127.0.0.1", 9100)) + .addLbEndpoints(endpoint("127.0.0.1", 9101))) + .removeEndpoints(0) + .build(); + checkEndpointsViaDiscoveryRequest(dogma.httpClient().uri(), expected, clusterName); + + // Deregister the original endpoint from the YAML-backed file. + final LocalityLbEndpoint toDeregister = + LocalityLbEndpoint.newBuilder().setLocality(locality) + .setLbEndpoint(endpoint("127.0.0.1", 9100)) + .build(); + final AggregatedHttpResponse deregisterResponse = + registerOrDeregister(endpointName, toDeregister, false); + assertOk(deregisterResponse); + + final ClusterLoadAssignment afterDeregister = + ClusterLoadAssignment.newBuilder() + .setClusterName(clusterName) + .addEndpoints(LocalityLbEndpoints.newBuilder() + .setLocality(locality) + .addLbEndpoints( + endpoint("127.0.0.1", 9101))) + .build(); + checkEndpointsViaDiscoveryRequest(dogma.httpClient().uri(), afterDeregister, clusterName); + } } diff --git a/xds/src/test/java/com/linecorp/centraldogma/xds/internal/XdsEndpointReadServiceTest.java b/xds/src/test/java/com/linecorp/centraldogma/xds/internal/XdsEndpointReadServiceTest.java new file mode 100644 index 000000000..04f98ef3b --- /dev/null +++ b/xds/src/test/java/com/linecorp/centraldogma/xds/internal/XdsEndpointReadServiceTest.java @@ -0,0 +1,265 @@ +/* + * Copyright 2026 LY Corporation + * + * LY Corporation 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: + * + * https://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. + */ +package com.linecorp.centraldogma.xds.internal; + +import static com.linecorp.centraldogma.xds.internal.ControlPlanePlugin.XDS_CENTRAL_DOGMA_PROJECT; +import static com.linecorp.centraldogma.xds.internal.ControlPlaneService.ENDPOINTS_DIRECTORY; +import static com.linecorp.centraldogma.xds.internal.ControlPlaneService.K8S_ENDPOINTS_DIRECTORY; +import static com.linecorp.centraldogma.xds.internal.XdsResourceManager.JSON_MESSAGE_MARSHALLER; +import static com.linecorp.centraldogma.xds.internal.XdsTestUtil.createEndpoint; +import static com.linecorp.centraldogma.xds.internal.XdsTestUtil.createGroup; +import static com.linecorp.centraldogma.xds.internal.XdsTestUtil.loadAssignment; +import static org.assertj.core.api.Assertions.assertThat; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import com.fasterxml.jackson.databind.JsonNode; + +import com.linecorp.armeria.common.AggregatedHttpResponse; +import com.linecorp.armeria.common.HttpStatus; +import com.linecorp.centraldogma.common.Change; +import com.linecorp.centraldogma.internal.Jackson; +import com.linecorp.centraldogma.testing.junit.CentralDogmaExtension; + +import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment; + +class XdsEndpointReadServiceTest { + + private static final String GROUP = "read-svc"; + + @RegisterExtension + static final CentralDogmaExtension dogma = new CentralDogmaExtension(); + + @BeforeAll + static void setup() { + assertThat(createGroup(GROUP, dogma.httpClient()).status()).isSameAs(HttpStatus.OK); + } + + // ---- listEndpoints ------------------------------------------------------- + + @Test + void listEndpoints_jsonFileAppearsWithJsonType() throws Exception { + final ClusterLoadAssignment endpoint = + loadAssignment("groups/" + GROUP + "/endpoints/list-json", "127.0.0.1", 8080); + assertThat(createEndpoint("groups/" + GROUP, "list-json", endpoint, dogma.httpClient()) + .status()).isSameAs(HttpStatus.OK); + + final AggregatedHttpResponse response = listEndpoints(); + assertThat(response.status()).isSameAs(HttpStatus.OK); + + final JsonNode body = Jackson.readTree(response.contentUtf8()); + final JsonNode entry = findEntryByPath(body, "/endpoints/list-json.json"); + assertThat(entry).isNotNull(); + assertThat(entry.get("type").asText()).isEqualTo("JSON"); + assertThat(entry.get("revision").asInt()).isPositive(); + } + + @Test + void listEndpoints_yamlFileAppearsWithYamlType() throws Exception { + final ClusterLoadAssignment endpoint = + loadAssignment("groups/" + GROUP + "/endpoints/list-yaml", "127.0.0.1", 8081); + pushYamlEndpoint("list-yaml", endpoint); + + final AggregatedHttpResponse response = listEndpoints(); + assertThat(response.status()).isSameAs(HttpStatus.OK); + + final JsonNode body = Jackson.readTree(response.contentUtf8()); + final JsonNode entry = findEntryByPath(body, "/endpoints/list-yaml.yaml"); + assertThat(entry).isNotNull(); + assertThat(entry.get("type").asText()).isEqualTo("YAML"); + assertThat(entry.get("revision").asInt()).isPositive(); + } + + @Test + void listEndpoints_mixedJsonAndYamlBothAppear() throws Exception { + final ClusterLoadAssignment jsonEndpoint = + loadAssignment("groups/" + GROUP + "/endpoints/list-mix-json", "127.0.0.1", 8082); + assertThat(createEndpoint("groups/" + GROUP, "list-mix-json", jsonEndpoint, dogma.httpClient()) + .status()).isSameAs(HttpStatus.OK); + + final ClusterLoadAssignment yamlEndpoint = + loadAssignment("groups/" + GROUP + "/endpoints/list-mix-yaml", "127.0.0.1", 8083); + pushYamlEndpoint("list-mix-yaml", yamlEndpoint); + + final AggregatedHttpResponse response = listEndpoints(); + assertThat(response.status()).isSameAs(HttpStatus.OK); + + final JsonNode body = Jackson.readTree(response.contentUtf8()); + assertThat(findEntryByPath(body, "/endpoints/list-mix-json.json")).isNotNull(); + assertThat(findEntryByPath(body, "/endpoints/list-mix-yaml.yaml")).isNotNull(); + } + + @Test + void listEndpoints_includesK8sJsonAndYamlEndpoints() throws Exception { + pushJsonK8sEndpoint("list-k8s-json", + loadAssignment("groups/" + GROUP + "/clusters/list-k8s-json", + "10.0.0.1", 9090)); + pushYamlK8sEndpoint("list-k8s-yaml", + loadAssignment("groups/" + GROUP + "/clusters/list-k8s-yaml", + "10.0.0.2", 9091)); + + final AggregatedHttpResponse response = listEndpoints(); + assertThat(response.status()).isSameAs(HttpStatus.OK); + + final JsonNode body = Jackson.readTree(response.contentUtf8()); + final JsonNode jsonEntry = findEntryByPath(body, "/k8s/endpoints/list-k8s-json.json"); + assertThat(jsonEntry).isNotNull(); + assertThat(jsonEntry.get("type").asText()).isEqualTo("JSON"); + + final JsonNode yamlEntry = findEntryByPath(body, "/k8s/endpoints/list-k8s-yaml.yaml"); + assertThat(yamlEntry).isNotNull(); + assertThat(yamlEntry.get("type").asText()).isEqualTo("YAML"); + } + + // ---- getEndpoint --------------------------------------------------------- + + @Test + void getEndpoint_jsonFileReturnsContentWithJsonType() throws Exception { + final ClusterLoadAssignment endpoint = + loadAssignment("groups/" + GROUP + "/endpoints/get-json", "127.0.0.1", 8090); + assertThat(createEndpoint("groups/" + GROUP, "get-json", endpoint, dogma.httpClient()) + .status()).isSameAs(HttpStatus.OK); + + final AggregatedHttpResponse response = getEndpoint("get-json"); + assertThat(response.status()).isSameAs(HttpStatus.OK); + + final JsonNode body = Jackson.readTree(response.contentUtf8()); + assertThat(body.get("path").asText()).isEqualTo("/endpoints/get-json.json"); + assertThat(body.get("type").asText()).isEqualTo("JSON"); + assertThat(body.get("revision").asInt()).isPositive(); + // Verify the endpoint content is present. + assertThat(body.get("content")).isNotNull(); + assertThat(body.get("content").toString()).contains("127.0.0.1"); + } + + @Test + void getEndpoint_yamlFileReturnsContentWithYamlType() throws Exception { + final ClusterLoadAssignment endpoint = + loadAssignment("groups/" + GROUP + "/endpoints/get-yaml", "127.0.0.1", 8091); + pushYamlEndpoint("get-yaml", endpoint); + + final AggregatedHttpResponse response = getEndpoint("get-yaml"); + assertThat(response.status()).isSameAs(HttpStatus.OK); + + final JsonNode body = Jackson.readTree(response.contentUtf8()); + assertThat(body.get("path").asText()).isEqualTo("/endpoints/get-yaml.yaml"); + assertThat(body.get("type").asText()).isEqualTo("YAML"); + assertThat(body.get("revision").asInt()).isPositive(); + // Content must be the deserialized endpoint, regardless of the on-disk format. + assertThat(body.get("content")).isNotNull(); + assertThat(body.get("content").toString()).contains("127.0.0.1"); + } + + @Test + void getEndpoint_notFound() { + final AggregatedHttpResponse response = getEndpoint("does-not-exist"); + assertThat(response.status()).isNotEqualTo(HttpStatus.OK); + } + + // ---- getK8sEndpoint ------------------------------------------------------ + + @Test + void getK8sEndpoint_jsonFileReturnsContentWithJsonType() throws Exception { + final ClusterLoadAssignment endpoint = + loadAssignment("groups/" + GROUP + "/clusters/get-k8s-json", "10.0.0.3", 9092); + pushJsonK8sEndpoint("get-k8s-json", endpoint); + + final AggregatedHttpResponse response = getK8sEndpoint("get-k8s-json"); + assertThat(response.status()).isSameAs(HttpStatus.OK); + + final JsonNode body = Jackson.readTree(response.contentUtf8()); + assertThat(body.get("path").asText()).isEqualTo("/k8s/endpoints/get-k8s-json.json"); + assertThat(body.get("type").asText()).isEqualTo("JSON"); + assertThat(body.get("content").toString()).contains("10.0.0.3"); + } + + @Test + void getK8sEndpoint_yamlFileReturnsContentWithYamlType() throws Exception { + final ClusterLoadAssignment endpoint = + loadAssignment("groups/" + GROUP + "/clusters/get-k8s-yaml", "10.0.0.4", 9093); + pushYamlK8sEndpoint("get-k8s-yaml", endpoint); + + final AggregatedHttpResponse response = getK8sEndpoint("get-k8s-yaml"); + assertThat(response.status()).isSameAs(HttpStatus.OK); + + final JsonNode body = Jackson.readTree(response.contentUtf8()); + assertThat(body.get("path").asText()).isEqualTo("/k8s/endpoints/get-k8s-yaml.yaml"); + assertThat(body.get("type").asText()).isEqualTo("YAML"); + assertThat(body.get("content").toString()).contains("10.0.0.4"); + } + + // ---- helpers ------------------------------------------------------------- + + private static AggregatedHttpResponse listEndpoints() { + return dogma.httpClient() + .get("/api/v1/xds/groups/" + GROUP + "/endpoints") + .aggregate().join(); + } + + private static AggregatedHttpResponse getEndpoint(String id) { + return dogma.httpClient() + .get("/api/v1/xds/groups/" + GROUP + "/endpoints/" + id) + .aggregate().join(); + } + + private static AggregatedHttpResponse getK8sEndpoint(String id) { + return dogma.httpClient() + .get("/api/v1/xds/groups/" + GROUP + "/k8s/endpoints/" + id) + .aggregate().join(); + } + + private static void pushYamlEndpoint(String endpointId, ClusterLoadAssignment endpoint) + throws Exception { + dogma.client().forRepo(XDS_CENTRAL_DOGMA_PROJECT, GROUP) + .commit("Add YAML endpoint: " + endpointId, + Change.ofYamlUpsert(ENDPOINTS_DIRECTORY + endpointId + ".yaml", + JSON_MESSAGE_MARSHALLER.writeValueAsString(endpoint))) + .push().join(); + } + + private static void pushJsonK8sEndpoint(String endpointId, ClusterLoadAssignment endpoint) + throws Exception { + dogma.client().forRepo(XDS_CENTRAL_DOGMA_PROJECT, GROUP) + .commit("Add JSON k8s endpoint: " + endpointId, + Change.ofJsonUpsert(K8S_ENDPOINTS_DIRECTORY + endpointId + ".json", + Jackson.readTree( + JSON_MESSAGE_MARSHALLER.writeValueAsString(endpoint)))) + .push().join(); + } + + private static void pushYamlK8sEndpoint(String endpointId, ClusterLoadAssignment endpoint) + throws Exception { + dogma.client().forRepo(XDS_CENTRAL_DOGMA_PROJECT, GROUP) + .commit("Add YAML k8s endpoint: " + endpointId, + Change.ofYamlUpsert(K8S_ENDPOINTS_DIRECTORY + endpointId + ".yaml", + JSON_MESSAGE_MARSHALLER.writeValueAsString(endpoint))) + .push().join(); + } + + /** + * Finds the array element whose {@code "path"} field equals {@code path}, or returns {@code null}. + */ + private static JsonNode findEntryByPath(JsonNode array, String path) { + for (JsonNode element : array) { + if (path.equals(element.get("path").asText())) { + return element; + } + } + return null; + } +} diff --git a/xds/src/test/java/com/linecorp/centraldogma/xds/internal/XdsResourceWatchingServiceTest.java b/xds/src/test/java/com/linecorp/centraldogma/xds/internal/XdsResourceWatchingServiceTest.java index 82862d56e..931e817e7 100644 --- a/xds/src/test/java/com/linecorp/centraldogma/xds/internal/XdsResourceWatchingServiceTest.java +++ b/xds/src/test/java/com/linecorp/centraldogma/xds/internal/XdsResourceWatchingServiceTest.java @@ -18,14 +18,19 @@ import static org.assertj.core.api.Assertions.assertThat; import java.io.IOException; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; +import com.fasterxml.jackson.databind.JsonNode; + import com.linecorp.centraldogma.client.CentralDogma; import com.linecorp.centraldogma.common.Change; import com.linecorp.centraldogma.server.storage.project.Project; @@ -40,6 +45,143 @@ class XdsResourceWatchingServiceTest { private static final BlockingQueue queue = new LinkedBlockingQueue<>(); + @Test + void yamlFilesAreHandledLikeJson() throws InterruptedException { + final BlockingQueue localQueue = new LinkedBlockingQueue<>(); + final CentralDogma client = dogma.client(); + client.createProject("yaml-watch-test").join(); + client.createRepository("yaml-watch-test", "repo").join(); + final Project project = dogma.projectManager().get("yaml-watch-test"); + + final XdsResourceWatchingService svc = new XdsResourceWatchingService( + project, "xds.", Metrics.globalRegistry) { + + private final ScheduledExecutorService exec = Executors.newSingleThreadScheduledExecutor(); + + @Override + protected ScheduledExecutorService executor() { + return exec; + } + + @Override + protected String pathPattern() { + return "/**"; + } + + @Override + protected void handleXdsResource(String path, JsonNode content, String groupName) { + localQueue.add("handleXdsResource: " + path); + localQueue.add("content.key=" + content.path("key").asText()); + } + + @Override + protected void onGroupRemoved(String groupName) { + localQueue.add(groupName + " removed"); + } + + @Override + protected void onFileRemoved(String groupName, String path) { + localQueue.add(path + " removed"); + } + + @Override + protected void onDiffHandled(String groupName) { + localQueue.add("diff handled: " + groupName); + } + + @Override + protected boolean isStopped() { + return false; + } + }; + svc.init(); + + // YAML upsert should trigger handleXdsResource with the content parsed as JsonNode. + client.forRepo("yaml-watch-test", "repo") + .commit("Add YAML file", Change.ofYamlUpsert("/c.yaml", "key: value")) + .push().join(); + assertThat(localQueue.take()).isEqualTo("handleXdsResource: /c.yaml"); + assertThat(localQueue.take()).isEqualTo("content.key=value"); + assertThat(localQueue.take()).isEqualTo("diff handled: repo"); + + // Updating an existing YAML file should also pass updated content as JsonNode. + client.forRepo("yaml-watch-test", "repo") + .commit("Update YAML file", Change.ofYamlUpsert("/c.yaml", "key: updated")) + .push().join(); + assertThat(localQueue.take()).isEqualTo("handleXdsResource: /c.yaml"); + assertThat(localQueue.take()).isEqualTo("content.key=updated"); + assertThat(localQueue.take()).isEqualTo("diff handled: repo"); + + // Removing a YAML file should trigger onFileRemoved. + client.forRepo("yaml-watch-test", "repo") + .commit("Remove YAML file", Change.ofRemoval("/c.yaml")) + .push().join(); + assertThat(localQueue.take()).isEqualTo("/c.yaml removed"); + assertThat(localQueue.take()).isEqualTo("diff handled: repo"); + } + + @Test + void yamlFilesLoadedDuringInit() throws InterruptedException { + final BlockingQueue localQueue = new LinkedBlockingQueue<>(); + final CentralDogma client = dogma.client(); + client.createProject("yaml-init-test").join(); + client.createRepository("yaml-init-test", "repo").join(); + + // Push YAML files BEFORE calling init() so they are picked up during the initial scan. + client.forRepo("yaml-init-test", "repo") + .commit("Seed YAML files", + Change.ofYamlUpsert("/x.yaml", "a: 1"), + Change.ofYamlUpsert("/y.yaml", "b: 2")) + .push().join(); + + final Project project = dogma.projectManager().get("yaml-init-test"); + final XdsResourceWatchingService svc = new XdsResourceWatchingService( + project, "xds.", Metrics.globalRegistry) { + + private final ScheduledExecutorService exec = Executors.newSingleThreadScheduledExecutor(); + + @Override + protected ScheduledExecutorService executor() { + return exec; + } + + @Override + protected String pathPattern() { + return "/**"; + } + + @Override + protected void handleXdsResource(String path, JsonNode content, String groupName) { + // Record path and its first field value to verify YAML content arrives as JsonNode. + final String firstValue = + content.fields().hasNext() ? content.fields().next().getValue().asText() : ""; + localQueue.add("handleXdsResource: " + path + " val=" + firstValue); + } + + @Override + protected void onGroupRemoved(String groupName) {} + + @Override + protected void onFileRemoved(String groupName, String path) {} + + @Override + protected void onDiffHandled(String groupName) {} + + @Override + protected boolean isStopped() { + return false; + } + }; + svc.init(); + + // Both YAML files should have been loaded during init with their content as JsonNode. + final List loaded = new ArrayList<>(); + loaded.add(localQueue.poll(2, TimeUnit.SECONDS)); + loaded.add(localQueue.poll(2, TimeUnit.SECONDS)); + assertThat(loaded).containsExactlyInAnyOrder("handleXdsResource: /x.yaml val=1", + "handleXdsResource: /y.yaml val=2"); + } + @Test void foo() throws InterruptedException { final CentralDogma client = dogma.client(); @@ -91,7 +233,7 @@ protected String pathPattern() { } @Override - protected void handleXdsResource(String path, String contentAsText, String groupName) + protected void handleXdsResource(String path, JsonNode content, String groupName) throws IOException { queue.add("handleXdsResource: " + path); } diff --git a/xds/src/test/java/com/linecorp/centraldogma/xds/internal/XdsYamlCompatibilityTest.java b/xds/src/test/java/com/linecorp/centraldogma/xds/internal/XdsYamlCompatibilityTest.java new file mode 100644 index 000000000..d48d2b596 --- /dev/null +++ b/xds/src/test/java/com/linecorp/centraldogma/xds/internal/XdsYamlCompatibilityTest.java @@ -0,0 +1,281 @@ +/* + * Copyright 2026 LY Corporation + * + * LY Corporation 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: + * + * https://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. + */ +package com.linecorp.centraldogma.xds.internal; + +import static com.linecorp.centraldogma.xds.internal.ControlPlanePlugin.XDS_CENTRAL_DOGMA_PROJECT; +import static com.linecorp.centraldogma.xds.internal.ControlPlaneService.CLUSTERS_DIRECTORY; +import static com.linecorp.centraldogma.xds.internal.ControlPlaneService.ENDPOINTS_DIRECTORY; +import static com.linecorp.centraldogma.xds.internal.XdsResourceManager.JSON_MESSAGE_MARSHALLER; +import static com.linecorp.centraldogma.xds.internal.XdsTestUtil.cluster; +import static com.linecorp.centraldogma.xds.internal.XdsTestUtil.createGroup; +import static com.linecorp.centraldogma.xds.internal.XdsTestUtil.loadAssignment; +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +import java.util.List; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import com.fasterxml.jackson.databind.JsonNode; +import com.google.protobuf.Any; +import com.google.protobuf.Duration; +import com.google.protobuf.InvalidProtocolBufferException; + +import com.linecorp.armeria.client.grpc.GrpcClients; +import com.linecorp.armeria.common.AggregatedHttpResponse; +import com.linecorp.armeria.common.HttpHeaderNames; +import com.linecorp.armeria.common.HttpMethod; +import com.linecorp.armeria.common.HttpStatus; +import com.linecorp.armeria.common.MediaType; +import com.linecorp.armeria.common.RequestHeaders; +import com.linecorp.centraldogma.common.Change; +import com.linecorp.centraldogma.common.Revision; +import com.linecorp.centraldogma.server.storage.repository.FindOptions; +import com.linecorp.centraldogma.server.storage.repository.Repository; +import com.linecorp.centraldogma.testing.junit.CentralDogmaExtension; + +import io.envoyproxy.controlplane.cache.Resources.V3; +import io.envoyproxy.envoy.config.cluster.v3.Cluster; +import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment; +import io.envoyproxy.envoy.service.cluster.v3.ClusterDiscoveryServiceGrpc.ClusterDiscoveryServiceStub; +import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest; +import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse; +import io.grpc.stub.StreamObserver; + +class XdsYamlCompatibilityTest { + + @RegisterExtension + static final CentralDogmaExtension dogma = new CentralDogmaExtension(); + + @BeforeAll + static void setup() { + assertThat(createGroup("foo", dogma.httpClient()).status()).isSameAs(HttpStatus.OK); + } + + @Test + void controlPlaneLoadsYamlCluster() throws Exception { + // Simulate a YAML file existing in the repo (as if it was migrated from JSON). + final String clusterName = "groups/foo/clusters/yaml-load-cluster"; + final Cluster cluster = cluster(clusterName, 1); + pushYamlCluster("yaml-load-cluster", cluster); + + // The control plane must serve the cluster read from the .yaml file. + await().pollInterval(100, TimeUnit.MILLISECONDS) + .untilAsserted(() -> checkClusterViaDiscovery(clusterName, cluster, true)); + } + + @Test + void updateYamlCluster() throws Exception { + final String clusterName = "groups/foo/clusters/yaml-update-cluster"; + final Cluster cluster = cluster(clusterName, 1); + pushYamlCluster("yaml-update-cluster", cluster); + + await().pollInterval(100, TimeUnit.MILLISECONDS) + .untilAsserted(() -> checkClusterViaDiscovery(clusterName, cluster, true)); + + // Update via the gRPC/HTTP API — updateOrDelete must locate the .yaml file. + final Cluster updated = cluster.toBuilder() + .setConnectTimeout(Duration.newBuilder().setSeconds(2).build()) + .build(); + final AggregatedHttpResponse response = patchCluster("foo", "yaml-update-cluster", updated); + assertThat(response.status()).isSameAs(HttpStatus.OK); + assertThat(response.headers().get("grpc-status")).isEqualTo("0"); + + // The .yaml file must be updated in place; no new .json file should appear. + final Repository repo = xdsRepo("foo"); + assertThat(repo.find(Revision.HEAD, CLUSTERS_DIRECTORY + "yaml-update-cluster.yaml", + FindOptions.FIND_ONE_WITHOUT_CONTENT).join()).isNotEmpty(); + assertThat(repo.find(Revision.HEAD, CLUSTERS_DIRECTORY + "yaml-update-cluster.json", + FindOptions.FIND_ONE_WITHOUT_CONTENT).join()).isEmpty(); + + // Control plane must serve the updated cluster. + await().pollInterval(100, TimeUnit.MILLISECONDS) + .untilAsserted(() -> checkClusterViaDiscovery(clusterName, updated, true)); + } + + @Test + void deleteYamlCluster() throws Exception { + final String clusterName = "groups/foo/clusters/yaml-delete-cluster"; + final Cluster cluster = cluster(clusterName, 1); + pushYamlCluster("yaml-delete-cluster", cluster); + + await().pollInterval(100, TimeUnit.MILLISECONDS) + .untilAsserted(() -> checkClusterViaDiscovery(clusterName, cluster, true)); + + // Delete via the HTTP API — updateOrDelete must locate and remove the .yaml file. + final AggregatedHttpResponse response = deleteCluster(clusterName); + assertThat(response.status()).isSameAs(HttpStatus.OK); + assertThat(response.headers().get("grpc-status")).isEqualTo("0"); + assertThat(response.contentUtf8()).isEqualTo("{}"); + + // The .yaml file must be gone. + final Repository repo = xdsRepo("foo"); + assertThat(repo.find(Revision.HEAD, CLUSTERS_DIRECTORY + "yaml-delete-cluster.yaml", + FindOptions.FIND_ONE_WITHOUT_CONTENT).join()).isEmpty(); + + // Control plane must no longer serve the deleted cluster. + await().pollInterval(100, TimeUnit.MILLISECONDS) + .untilAsserted(() -> checkClusterViaDiscovery(clusterName, cluster, false)); + } + + @Test + void endpointListIncludesYamlFiles() throws Exception { + final String endpointName = "groups/foo/endpoints/yaml-list-endpoint"; + final ClusterLoadAssignment endpoint = loadAssignment(endpointName, "127.0.0.1", 8080); + pushYamlEndpoint("yaml-list-endpoint", endpoint); + + // listEndpoints must include the YAML endpoint with type "YAML". + await().pollInterval(100, TimeUnit.MILLISECONDS).untilAsserted(() -> { + final AggregatedHttpResponse response = + dogma.httpClient().get("/api/v1/xds/groups/foo/endpoints").aggregate().join(); + assertThat(response.status()).isSameAs(HttpStatus.OK); + assertThat(response.contentUtf8()).contains("/endpoints/yaml-list-endpoint.yaml"); + assertThat(response.contentUtf8()).contains("\"type\":\"YAML\""); + }); + } + + @Test + void endpointGetReturnsYamlContent() throws Exception { + final String endpointName = "groups/foo/endpoints/yaml-get-endpoint"; + final ClusterLoadAssignment endpoint = loadAssignment(endpointName, "127.0.0.1", 8081); + pushYamlEndpoint("yaml-get-endpoint", endpoint); + + // getEndpoint must find and return the content of the .yaml file. + await().pollInterval(100, TimeUnit.MILLISECONDS).untilAsserted(() -> { + final AggregatedHttpResponse response = + dogma.httpClient().get("/api/v1/xds/groups/foo/endpoints/yaml-get-endpoint") + .aggregate().join(); + assertThat(response.status()).isSameAs(HttpStatus.OK); + final JsonNode body = com.linecorp.centraldogma.internal.Jackson.readTree(response.contentUtf8()); + assertThat(body.get("path").asText()).isEqualTo("/endpoints/yaml-get-endpoint.yaml"); + assertThat(body.get("type").asText()).isEqualTo("YAML"); + assertThat(response.contentUtf8()).contains("127.0.0.1"); + }); + } + + @Test + void jsonFileStillWorksAlongsideYaml() throws Exception { + // Push a JSON cluster via the normal API (still .json). + final String jsonClusterName = "groups/foo/clusters/json-alongside-yaml"; + final AggregatedHttpResponse createResp = createCluster("foo", "json-alongside-yaml", + cluster(jsonClusterName, 1)); + assertThat(createResp.status()).isSameAs(HttpStatus.OK); + + // Push a YAML cluster directly. + final String yamlClusterName = "groups/foo/clusters/yaml-alongside-json"; + final Cluster yamlCluster = cluster(yamlClusterName, 1); + pushYamlCluster("yaml-alongside-json", yamlCluster); + + // Both must be served by the control plane. + final Cluster expectedJson = cluster(jsonClusterName, 1).toBuilder() + .setRespectDnsTtl(true).build(); + await().pollInterval(100, TimeUnit.MILLISECONDS).untilAsserted(() -> { + checkClusterViaDiscovery(jsonClusterName, expectedJson, true); + checkClusterViaDiscovery(yamlClusterName, yamlCluster, true); + }); + } + + // ---- helpers ---- + + private static void pushYamlCluster(String clusterId, Cluster cluster) throws Exception { + final String content = JSON_MESSAGE_MARSHALLER.writeValueAsString(cluster); + dogma.client().forRepo(XDS_CENTRAL_DOGMA_PROJECT, "foo") + .commit("Add YAML cluster: " + clusterId, + Change.ofYamlUpsert(CLUSTERS_DIRECTORY + clusterId + ".yaml", content)) + .push().join(); + } + + private static void pushYamlEndpoint(String endpointId, ClusterLoadAssignment endpoint) + throws Exception { + final String content = JSON_MESSAGE_MARSHALLER.writeValueAsString(endpoint); + dogma.client().forRepo(XDS_CENTRAL_DOGMA_PROJECT, "foo") + .commit("Add YAML endpoint: " + endpointId, + Change.ofYamlUpsert(ENDPOINTS_DIRECTORY + endpointId + ".yaml", content)) + .push().join(); + } + + private static AggregatedHttpResponse createCluster(String group, String clusterId, Cluster cluster) + throws Exception { + return dogma.httpClient() + .prepare() + .method(HttpMethod.POST) + .path("/api/v1/xds/groups/" + group + "/clusters") + .queryParam("cluster_id", clusterId) + .header(HttpHeaderNames.AUTHORIZATION, "Bearer anonymous") + .content(MediaType.JSON_UTF_8, JSON_MESSAGE_MARSHALLER.writeValueAsString(cluster)) + .execute().aggregate().join(); + } + + private static AggregatedHttpResponse patchCluster(String group, String clusterId, Cluster cluster) + throws Exception { + final RequestHeaders headers = + RequestHeaders.builder(HttpMethod.PATCH, + "/api/v1/xds/groups/" + group + "/clusters/" + clusterId) + .set(HttpHeaderNames.AUTHORIZATION, "Bearer anonymous") + .contentType(MediaType.JSON_UTF_8).build(); + return dogma.httpClient() + .execute(headers, JSON_MESSAGE_MARSHALLER.writeValueAsString(cluster)) + .aggregate().join(); + } + + private static AggregatedHttpResponse deleteCluster(String clusterName) { + final RequestHeaders headers = + RequestHeaders.builder(HttpMethod.DELETE, "/api/v1/xds/" + clusterName) + .set(HttpHeaderNames.AUTHORIZATION, "Bearer anonymous").build(); + return dogma.httpClient().execute(headers).aggregate().join(); + } + + private static Repository xdsRepo(String group) { + return dogma.projectManager().get(XDS_CENTRAL_DOGMA_PROJECT).repos().get(group); + } + + private static void checkClusterViaDiscovery(String clusterName, Cluster expectedCluster, + boolean shouldExist) + throws InterruptedException, InvalidProtocolBufferException { + final ClusterDiscoveryServiceStub client = + GrpcClients.newClient(dogma.httpClient().uri(), ClusterDiscoveryServiceStub.class); + final BlockingQueue queue = new ArrayBlockingQueue<>(2); + final StreamObserver req = client.streamClusters(new StreamObserver<>() { + + @Override + public void onNext(DiscoveryResponse value) { + queue.add(value); + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() {} + }); + req.onNext(DiscoveryRequest.newBuilder().setTypeUrl(V3.CLUSTER_TYPE_URL) + .addResourceNames(clusterName).build()); + if (shouldExist) { + final DiscoveryResponse resp = queue.take(); + final List resources = resp.getResourcesList(); + assertThat(resources).hasSize(1); + assertThat(Cluster.parseFrom(resources.get(0).getValue())).isEqualTo(expectedCluster); + } else { + final DiscoveryResponse resp = queue.poll(300, TimeUnit.MILLISECONDS); + assertThat(resp).isNull(); + } + } +} From 8ba05b12426dc4b9aa226f033de49074dc378f30 Mon Sep 17 00:00:00 2001 From: minwoox Date: Thu, 2 Jul 2026 10:51:23 +0900 Subject: [PATCH 2/3] Address comment from AI --- .../xds/internal/XdsResourceManager.java | 30 +++++++++++++++++++ .../endpoint/v1/XdsEndpointServiceTest.java | 28 +++++++++++++++++ 2 files changed, 58 insertions(+) diff --git a/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceManager.java b/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceManager.java index f3713507c..c904a347a 100644 --- a/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceManager.java +++ b/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceManager.java @@ -180,6 +180,36 @@ private void checkWritePermission0(String group) { public void push( StreamObserver responseObserver, String group, String resourceName, String fileName, String summary, T resource, Author author, boolean create) { + if (create) { + // Before attempting to create, verify that neither the requested file nor its alternative + // extension (.json ↔ .yaml) already exists, so a migrated .yaml resource is not silently + // shadowed by a newly created .json file. + final Repository repository = xdsProject.repos().get(group); + final String altFileName = alternativeFileName(fileName); + repository.find(Revision.HEAD, fileName + ',' + altFileName, FIND_ONE_WITHOUT_CONTENT) + .handle((entries, cause) -> { + if (cause != null) { + responseObserver.onError(cause); + return null; + } + if (!entries.isEmpty()) { + responseObserver.onError( + Status.ALREADY_EXISTS + .withDescription("Resource already exists: " + resourceName) + .asRuntimeException()); + return null; + } + doPush(responseObserver, group, resourceName, fileName, summary, resource, author, true); + return null; + }); + return; + } + doPush(responseObserver, group, resourceName, fileName, summary, resource, author, false); + } + + private void doPush( + StreamObserver responseObserver, String group, String resourceName, String fileName, + String summary, T resource, Author author, boolean create) { final Change change; try { final String jsonText = JSON_MESSAGE_MARSHALLER.writeValueAsString(resource); diff --git a/xds/src/test/java/com/linecorp/centraldogma/xds/endpoint/v1/XdsEndpointServiceTest.java b/xds/src/test/java/com/linecorp/centraldogma/xds/endpoint/v1/XdsEndpointServiceTest.java index 41d885fb0..7a652e5d3 100644 --- a/xds/src/test/java/com/linecorp/centraldogma/xds/endpoint/v1/XdsEndpointServiceTest.java +++ b/xds/src/test/java/com/linecorp/centraldogma/xds/endpoint/v1/XdsEndpointServiceTest.java @@ -61,6 +61,7 @@ import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest; import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse; import io.envoyproxy.envoy.service.endpoint.v3.EndpointDiscoveryServiceGrpc.EndpointDiscoveryServiceStub; +import io.grpc.Status; import io.grpc.stub.StreamObserver; public class XdsEndpointServiceTest { @@ -176,6 +177,33 @@ void updateEndpointViaHttp() throws Exception { checkEndpointsViaDiscoveryRequest(dogma.httpClient().uri(), actualEndpoint2, clusterName); } + @Test + void createEndpointReturnAlreadyExistsWhenYamlExists() throws Exception { + // Pre-populate the repo with a YAML endpoint (simulating a JSON→YAML migration). + final String clusterName = "groups/foo/clusters/yaml-exists/1"; + final ClusterLoadAssignment initial = loadAssignment(clusterName, "127.0.0.1", 8080); + dogma.client().forRepo(XDS_CENTRAL_DOGMA_PROJECT, "foo") + .commit("Add YAML endpoint", + Change.ofYamlUpsert(ENDPOINTS_DIRECTORY + "yaml-exists/1.yaml", + JSON_MESSAGE_MARSHALLER.writeValueAsString(initial))) + .push().join(); + + // A create request for the same logical resource must return ALREADY_EXISTS, not succeed. + final AggregatedHttpResponse response = + createEndpoint("groups/foo", "yaml-exists/1", initial, dogma.httpClient()); + assertThat(response.status()).isSameAs(HttpStatus.CONFLICT); + assertThat(response.headers().get("grpc-status")) + .isEqualTo(Integer.toString(Status.ALREADY_EXISTS.getCode().value())); + + // The original .yaml file must still be the only file present (no new .json created). + final Repository repo = + dogma.projectManager().get(XDS_CENTRAL_DOGMA_PROJECT).repos().get("foo"); + assertThat(repo.find(Revision.HEAD, ENDPOINTS_DIRECTORY + "yaml-exists/1.yaml", + FindOptions.FIND_ONE_WITHOUT_CONTENT).join()).isNotEmpty(); + assertThat(repo.find(Revision.HEAD, ENDPOINTS_DIRECTORY + "yaml-exists/1.json", + FindOptions.FIND_ONE_WITHOUT_CONTENT).join()).isEmpty(); + } + @Test void updateYamlEndpointViaHttp() throws Exception { // Push an endpoint as YAML directly (simulating a JSON→YAML migration). From 5afe8fd068eed1515d607e26fb846c1e47bf398d Mon Sep 17 00:00:00 2001 From: minwoox Date: Thu, 2 Jul 2026 17:27:36 +0900 Subject: [PATCH 3/3] Address the comment from @ikhoon --- .../xds/internal/XdsResourceManager.java | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceManager.java b/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceManager.java index c904a347a..810c6e16c 100644 --- a/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceManager.java +++ b/xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceManager.java @@ -23,7 +23,7 @@ import java.io.IOException; import java.lang.reflect.Method; import java.lang.reflect.Modifier; -import java.util.function.Function; +import java.util.function.Consumer; import java.util.regex.Pattern; import org.curioswitch.common.protobuf.json.MessageMarshaller; @@ -270,8 +270,8 @@ public void update(StreamObserver responseObserver, Strin String resourceName, String fileName, String summary, T resource, Author author) { updateOrDelete(responseObserver, group, resourceName, fileName, resolvedFileName -> - () -> push(responseObserver, group, resourceName, resolvedFileName, - summary, resource, author, false)); + push(responseObserver, group, resourceName, resolvedFileName, + summary, resource, author, false)); } public void delete(StreamObserver responseObserver, String group, @@ -281,7 +281,7 @@ public void delete(StreamObserver responseObserver, String group, public void delete(StreamObserver responseObserver, String group, String resourceName, String fileName, String summary, Author author) { - updateOrDelete(responseObserver, group, resourceName, fileName, resolvedFileName -> () -> + updateOrDelete(responseObserver, group, resourceName, fileName, resolvedFileName -> commandExecutor.execute(Command.push(author, XDS_CENTRAL_DOGMA_PROJECT, group, Revision.HEAD, summary, "", Markup.PLAINTEXT, ImmutableList.of(Change.ofRemoval(resolvedFileName)))) @@ -298,7 +298,7 @@ public void delete(StreamObserver responseObserver, String group, } public void updateOrDelete(StreamObserver responseObserver, String group, String resourceName, - String fileName, Function taskProvider) { + String fileName, Consumer taskProvider) { final Repository repository = xdsProject.repos().get(group); // Search for both the requested filename and its alternative extension (.json ↔ .yaml) // to support files that may have been written in either format. @@ -316,7 +316,7 @@ public void updateOrDelete(StreamObserver responseObserver, String group, Str return null; } final String resolvedFileName = entries.keySet().iterator().next(); - taskProvider.apply(resolvedFileName).run(); + taskProvider.accept(resolvedFileName); return null; }); }