Skip to content

Commit 6680dc6

Browse files
minwooxminwoox
committed
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<String, Runnable> 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.
1 parent b0011fd commit 6680dc6

14 files changed

Lines changed: 915 additions & 49 deletions

File tree

‎server/src/main/java/com/linecorp/centraldogma/server/command/ContentTransformer.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,8 @@ public class ContentTransformer<T> {
4040
*/
4141
public ContentTransformer(String path, EntryType entryType, BiFunction<Revision, T, T> transformer) {
4242
this.path = requireNonNull(path, "path");
43-
checkArgument(entryType == EntryType.JSON, "entryType: %s (expected: %s)", entryType, EntryType.JSON);
43+
checkArgument(entryType == EntryType.JSON || entryType == EntryType.YAML,
44+
"entryType: %s (expected: JSON or YAML)", entryType);
4445
this.entryType = requireNonNull(entryType, "entryType");
4546
this.transformer = requireNonNull(transformer, "transformer");
4647
}

‎server/src/main/java/com/linecorp/centraldogma/server/internal/storage/repository/git/TransformingChangesApplier.java‎

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -35,15 +35,17 @@
3535
import com.linecorp.centraldogma.common.EntryType;
3636
import com.linecorp.centraldogma.common.Revision;
3737
import com.linecorp.centraldogma.internal.Jackson;
38+
import com.linecorp.centraldogma.internal.Yaml;
3839
import com.linecorp.centraldogma.server.command.ContentTransformer;
3940

4041
final class TransformingChangesApplier extends AbstractChangesApplier {
4142

4243
private final ContentTransformer<JsonNode> transformer;
4344

4445
TransformingChangesApplier(ContentTransformer<?> transformer) {
45-
checkArgument(transformer.entryType() == EntryType.JSON,
46-
"transformer: %s (expected: JSON type)", transformer);
46+
checkArgument(transformer.entryType() == EntryType.JSON ||
47+
transformer.entryType() == EntryType.YAML,
48+
"transformer: %s (expected: JSON or YAML type)", transformer);
4749
//noinspection unchecked
4850
this.transformer = (ContentTransformer<JsonNode>) transformer;
4951
}
@@ -55,13 +57,20 @@ int doApply(Revision headRevision, DirCache dirCache,
5557
final DirCacheEntry oldEntry = dirCache.getEntry(changePath);
5658
final byte[] oldContent = oldEntry != null ? reader.open(oldEntry.getObjectId()).getBytes()
5759
: null;
58-
final JsonNode oldJsonNode = oldContent != null ? Jackson.readTree(oldContent)
60+
final boolean isYaml = transformer.entryType() == EntryType.YAML;
61+
final JsonNode oldJsonNode = oldContent != null ? (isYaml ? Yaml.readTree(oldContent)
62+
: Jackson.readTree(oldContent))
5963
: JsonNodeFactory.instance.nullNode();
6064
try {
6165
final JsonNode newJsonNode = transformer.transformer().apply(headRevision, oldJsonNode.deepCopy());
6266
requireNonNull(newJsonNode, "transformer.transformer().apply() returned null");
6367
if (!Objects.equals(newJsonNode, oldJsonNode)) {
64-
applyPathEdit(dirCache, new InsertJson(changePath, inserter, newJsonNode));
68+
if (isYaml) {
69+
applyPathEdit(dirCache, new InsertText(changePath, inserter,
70+
Yaml.writeValueAsString(newJsonNode)));
71+
} else {
72+
applyPathEdit(dirCache, new InsertJson(changePath, inserter, newJsonNode));
73+
}
6574
return 1;
6675
}
6776
} catch (CentralDogmaException e) {

‎xds/src/main/java/com/linecorp/centraldogma/xds/endpoint/v1/XdsEndpointUpdateScheduler.java‎

Lines changed: 19 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,16 @@ int batchUpdateTaskSize() {
8181
return batchUpdateTasks.size();
8282
}
8383

84+
private static String alternativeFileName(String fileName) {
85+
if (fileName.endsWith(".json")) {
86+
return fileName.substring(0, fileName.length() - 5) + ".yaml";
87+
}
88+
if (fileName.endsWith(".yaml")) {
89+
return fileName.substring(0, fileName.length() - 5) + ".json";
90+
}
91+
return fileName;
92+
}
93+
8494
void schedule(String group, String endpointName, String fileName,
8595
LocalityLbEndpoint localityLbEndpoint, StreamObserver<?> streamObserver, boolean register) {
8696
final EndpointIdentifier identifier = EndpointIdentifier.of(localityLbEndpoint);
@@ -170,11 +180,10 @@ private void flush() {
170180
}
171181
}
172182

173-
final ContentTransformer<JsonNode> transformer = new ContentTransformer<>(
174-
fileName, EntryType.JSON, new BatchUpdateTransformer(toRegister, toDeregister));
175-
176183
final Repository repository = xdsResourceManager.xdsProject().repos().get(group);
177-
repository.find(Revision.HEAD, fileName, FIND_ONE_WITHOUT_CONTENT).handle((entries, cause) -> {
184+
final String altFileName = alternativeFileName(fileName);
185+
repository.find(Revision.HEAD, fileName + ',' + altFileName, FIND_ONE_WITHOUT_CONTENT)
186+
.handle((entries, cause) -> {
178187
if (cause != null) {
179188
copied.forEach(pendingUpdate -> pendingUpdate.streamObserver.onError(cause));
180189
return null;
@@ -186,6 +195,11 @@ private void flush() {
186195
copied.forEach(pendingUpdate -> pendingUpdate.streamObserver.onError(runtimeException));
187196
return null;
188197
}
198+
final String resolvedFileName = entries.keySet().iterator().next();
199+
final EntryType entryType =
200+
resolvedFileName.endsWith(".yaml") ? EntryType.YAML : EntryType.JSON;
201+
final ContentTransformer<JsonNode> transformer = new ContentTransformer<>(
202+
resolvedFileName, entryType, new BatchUpdateTransformer(toRegister, toDeregister));
189203
final String commitMessage =
190204
"Batch update for " + endpointName + " in group " + group + ": " +
191205
toRegister.size() + " register, " + toDeregister.size() + " deregister";
@@ -304,8 +318,7 @@ private static ClusterLoadAssignment.Builder toClusterLoadAssignmentBuilder(Json
304318
final Builder clusterLoadAssignmentBuilder =
305319
ClusterLoadAssignment.newBuilder();
306320
try {
307-
JSON_MESSAGE_MARSHALLER.mergeValue(Jackson.writeValueAsString(oldJsonNode),
308-
clusterLoadAssignmentBuilder);
321+
JSON_MESSAGE_MARSHALLER.mergeValue(oldJsonNode.traverse(), clusterLoadAssignmentBuilder);
309322
} catch (Throwable t) {
310323
// Should never reach here.
311324
throw new Error();

‎xds/src/main/java/com/linecorp/centraldogma/xds/internal/CentralDogmaXdsResources.java‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -103,7 +103,8 @@ void removeCluster(String groupName, String path) {
103103
}
104104

105105
private static String getResourceName(String groupName, String path) {
106-
return "groups/" + groupName + path.substring(0, path.length() - 5); // Remove .json
106+
// Remove .json or .yaml (both 5 chars)
107+
return "groups/" + groupName + path.substring(0, path.length() - 5);
107108
}
108109

109110
void removeEndpoint(String groupName, String path) {
@@ -112,11 +113,11 @@ void removeEndpoint(String groupName, String path) {
112113
if (groupEndpoints == null) {
113114
return;
114115
}
115-
// e.g. /endpoints/foo-cluster.json file with group foo -> groups/foo/clusters/foo-cluster
116-
// e.g. /k8s/endpoints/foo-cluster.json file with group foo -> groups/foo/k8s/clusters/foo-cluster
116+
// e.g. /endpoints/foo-cluster.json/.yaml file with group foo -> groups/foo/clusters/foo-cluster
117+
// e.g. /k8s/endpoints/foo-cluster.json/.yaml file with group foo -> groups/foo/k8s/clusters/foo-cluster
117118
final String clusterName =
118119
"groups/" + groupName +
119-
ENDPOINTS_PATTERN.matcher(path.substring(0, path.length() - 5) /* remove .json */)
120+
ENDPOINTS_PATTERN.matcher(path.substring(0, path.length() - 5) /* remove .json or .yaml */)
120121
.replaceFirst("/clusters/");
121122
endpointUpdated |= groupEndpoints.remove(clusterName) != null;
122123
}

‎xds/src/main/java/com/linecorp/centraldogma/xds/internal/ControlPlaneService.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -338,23 +338,23 @@ protected String pathPattern() {
338338
}
339339

340340
@Override
341-
protected void handleXdsResource(String path, String contentAsText, String groupName)
341+
protected void handleXdsResource(String path, JsonNode content, String groupName)
342342
throws IOException {
343343
if (path.startsWith(CLUSTERS_DIRECTORY)) {
344344
final Cluster.Builder builder = Cluster.newBuilder();
345-
JSON_MESSAGE_MARSHALLER.mergeValue(contentAsText, builder);
345+
JSON_MESSAGE_MARSHALLER.mergeValue(content.traverse(), builder);
346346
centralDogmaXdsResources.setCluster(groupName, builder.build());
347347
} else if (path.startsWith(ENDPOINTS_DIRECTORY) || path.startsWith(K8S_ENDPOINTS_DIRECTORY)) {
348348
final ClusterLoadAssignment.Builder builder = ClusterLoadAssignment.newBuilder();
349-
JSON_MESSAGE_MARSHALLER.mergeValue(contentAsText, builder);
349+
JSON_MESSAGE_MARSHALLER.mergeValue(content.traverse(), builder);
350350
centralDogmaXdsResources.setEndpoint(groupName, builder.build());
351351
} else if (path.startsWith(LISTENERS_DIRECTORY)) {
352352
final Listener.Builder builder = Listener.newBuilder();
353-
JSON_MESSAGE_MARSHALLER.mergeValue(contentAsText, builder);
353+
JSON_MESSAGE_MARSHALLER.mergeValue(content.traverse(), builder);
354354
centralDogmaXdsResources.setListener(groupName, builder.build());
355355
} else if (path.startsWith(ROUTES_DIRECTORY)) {
356356
final RouteConfiguration.Builder builder = RouteConfiguration.newBuilder();
357-
JSON_MESSAGE_MARSHALLER.mergeValue(contentAsText, builder);
357+
JSON_MESSAGE_MARSHALLER.mergeValue(content.traverse(), builder);
358358
centralDogmaXdsResources.setRoute(groupName, builder.build());
359359
} else {
360360
// ignore

‎xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsEndpointReadService.java‎

Lines changed: 17 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -56,12 +56,12 @@ public CompletableFuture<JsonNode> listEndpoints(@Param String group) {
5656
.thenApply(entries -> {
5757
final ArrayNode array = JsonNodeFactory.instance.arrayNode();
5858
for (Entry<?> entry : entries.values()) {
59-
if (entry.type() != EntryType.JSON) {
59+
if (entry.type() != EntryType.JSON && entry.type() != EntryType.YAML) {
6060
continue;
6161
}
6262
final ObjectNode node = array.addObject();
6363
node.put("path", entry.path());
64-
node.put("type", "JSON");
64+
node.put("type", entry.type().name());
6565
node.put("revision", entry.revision().major());
6666
}
6767
return array;
@@ -73,7 +73,7 @@ public CompletableFuture<JsonNode> listEndpoints(@Param String group) {
7373
*/
7474
@Get("/xds/groups/{group}/endpoints/{*id}")
7575
public CompletableFuture<JsonNode> getEndpoint(@Param String group, @Param String id) {
76-
return read(group, ENDPOINTS_DIRECTORY + id + ".json");
76+
return readByBase(group, ENDPOINTS_DIRECTORY + id);
7777
}
7878

7979
/**
@@ -82,17 +82,27 @@ public CompletableFuture<JsonNode> getEndpoint(@Param String group, @Param Strin
8282
*/
8383
@Get("/xds/groups/{group}/k8s/endpoints/{*id}")
8484
public CompletableFuture<JsonNode> getK8sEndpoint(@Param String group, @Param String id) {
85-
return read(group, K8S_ENDPOINTS_DIRECTORY + id + ".json");
85+
return readByBase(group, K8S_ENDPOINTS_DIRECTORY + id);
8686
}
8787

88-
private CompletableFuture<JsonNode> read(String group, String path) {
89-
return xdsProject.repos().get(group).get(Revision.HEAD, path).thenApply(XdsEndpointReadService::toNode);
88+
private CompletableFuture<JsonNode> readByBase(String group, String pathBase) {
89+
final Repository repository = xdsProject.repos().get(group);
90+
return repository.find(Revision.HEAD, pathBase + ".json," + pathBase + ".yaml")
91+
.thenCompose(entries -> {
92+
if (entries.isEmpty()) {
93+
// Delegate to get() so the caller receives a proper EntryNotFoundException.
94+
return repository.get(Revision.HEAD, pathBase + ".json")
95+
.thenApply(XdsEndpointReadService::toNode);
96+
}
97+
return CompletableFuture.completedFuture(
98+
toNode(entries.values().iterator().next()));
99+
});
90100
}
91101

92102
private static JsonNode toNode(Entry<?> entry) {
93103
final ObjectNode node = JsonNodeFactory.instance.objectNode();
94104
node.put("path", entry.path());
95-
node.put("type", "JSON");
105+
node.put("type", entry.type().name());
96106
node.put("revision", entry.revision().major());
97107
node.set("content", (JsonNode) entry.content());
98108
return node;

‎xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceManager.java‎

Lines changed: 26 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import java.io.IOException;
2424
import java.lang.reflect.Method;
2525
import java.lang.reflect.Modifier;
26+
import java.util.function.Function;
2627
import java.util.regex.Pattern;
2728

2829
import org.curioswitch.common.protobuf.json.MessageMarshaller;
@@ -185,6 +186,8 @@ public <T extends Message> void push(
185186
final JsonNode jsonNode = Jackson.readTree(jsonText);
186187
if (create) {
187188
change = Change.ofJsonPatch(fileName, null, jsonNode);
189+
} else if (fileName.endsWith(".yaml")) {
190+
change = Change.ofYamlUpsert(fileName, jsonNode);
188191
} else {
189192
change = Change.ofJsonUpsert(fileName, jsonNode);
190193
}
@@ -236,8 +239,8 @@ public <T extends Message> void update(StreamObserver<T> responseObserver, Strin
236239
public <T extends Message> void update(StreamObserver<T> responseObserver, String group,
237240
String resourceName, String fileName, String summary, T resource,
238241
Author author) {
239-
updateOrDelete(responseObserver, group, resourceName, fileName,
240-
() -> push(responseObserver, group, resourceName, fileName,
242+
updateOrDelete(responseObserver, group, resourceName, fileName, resolvedFileName ->
243+
() -> push(responseObserver, group, resourceName, resolvedFileName,
241244
summary, resource, author, false));
242245
}
243246

@@ -248,10 +251,10 @@ public void delete(StreamObserver<Empty> responseObserver, String group,
248251

249252
public void delete(StreamObserver<Empty> responseObserver, String group,
250253
String resourceName, String fileName, String summary, Author author) {
251-
final Runnable deleteTask = () ->
254+
updateOrDelete(responseObserver, group, resourceName, fileName, resolvedFileName -> () ->
252255
commandExecutor.execute(Command.push(author, XDS_CENTRAL_DOGMA_PROJECT, group,
253256
Revision.HEAD, summary, "", Markup.PLAINTEXT,
254-
ImmutableList.of(Change.ofRemoval(fileName))))
257+
ImmutableList.of(Change.ofRemoval(resolvedFileName))))
255258
.handle((unused, cause) -> {
256259
if (cause != null) {
257260
responseObserver.onError(
@@ -261,14 +264,17 @@ public void delete(StreamObserver<Empty> responseObserver, String group,
261264
responseObserver.onNext(Empty.getDefaultInstance());
262265
responseObserver.onCompleted();
263266
return null;
264-
});
265-
updateOrDelete(responseObserver, group, resourceName, fileName, deleteTask);
267+
}));
266268
}
267269

268270
public void updateOrDelete(StreamObserver<?> responseObserver, String group, String resourceName,
269-
String fileName, Runnable task) {
271+
String fileName, Function<String, Runnable> taskProvider) {
270272
final Repository repository = xdsProject.repos().get(group);
271-
repository.find(Revision.HEAD, fileName, FIND_ONE_WITHOUT_CONTENT).handle((entries, cause) -> {
273+
// Search for both the requested filename and its alternative extension (.json ↔ .yaml)
274+
// to support files that may have been written in either format.
275+
final String altFileName = alternativeFileName(fileName);
276+
repository.find(Revision.HEAD, fileName + ',' + altFileName, FIND_ONE_WITHOUT_CONTENT)
277+
.handle((entries, cause) -> {
272278
if (cause != null) {
273279
responseObserver.onError(cause);
274280
return null;
@@ -279,8 +285,19 @@ public void updateOrDelete(StreamObserver<?> responseObserver, String group, Str
279285
.asRuntimeException());
280286
return null;
281287
}
282-
task.run();
288+
final String resolvedFileName = entries.keySet().iterator().next();
289+
taskProvider.apply(resolvedFileName).run();
283290
return null;
284291
});
285292
}
293+
294+
private static String alternativeFileName(String fileName) {
295+
if (fileName.endsWith(".json")) {
296+
return fileName.substring(0, fileName.length() - 5) + ".yaml";
297+
}
298+
if (fileName.endsWith(".yaml")) {
299+
return fileName.substring(0, fileName.length() - 5) + ".json";
300+
}
301+
return fileName;
302+
}
286303
}

‎xds/src/main/java/com/linecorp/centraldogma/xds/internal/XdsResourceWatchingService.java‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import org.slf4j.Logger;
2727
import org.slf4j.LoggerFactory;
2828

29+
import com.fasterxml.jackson.databind.JsonNode;
2930
import com.google.common.collect.ImmutableList;
3031
import com.google.common.collect.ImmutableList.Builder;
3132
import com.google.common.collect.Sets;
@@ -70,7 +71,7 @@ protected Project xdsProject() {
7071

7172
protected abstract String pathPattern();
7273

73-
protected abstract void handleXdsResource(String path, String contentAsText, String groupName)
74+
protected abstract void handleXdsResource(String path, JsonNode content, String groupName)
7475
throws IOException;
7576

7677
protected abstract void onGroupRemoved(String groupName);
@@ -110,13 +111,13 @@ protected void init() {
110111
" at revision: " + normalizedRevision, cause);
111112
}
112113
for (Entry<?> entry : entries.values()) {
113-
if (entry.type() != EntryType.JSON || !entry.hasContent()) {
114+
if ((entry.type() != EntryType.JSON && entry.type() != EntryType.YAML) ||
115+
!entry.hasContent()) {
114116
continue;
115117
}
116118
final String path = entry.path();
117-
final String contentAsText = entry.contentAsText();
118119
try {
119-
handleXdsResource(path, contentAsText, groupName);
120+
handleXdsResource(path, (JsonNode) entry.content(), groupName);
120121
} catch (Throwable t) {
121122
logger.warn("Unexpected exception while building an xDS resource from {}.",
122123
groupName + path, t);
@@ -222,8 +223,9 @@ private void handleDiff(String groupName, Revision newRevision,
222223
final String path = change.path();
223224
switch (change.type()) {
224225
case UPSERT_JSON:
226+
case UPSERT_YAML:
225227
try {
226-
handleXdsResource(path, change.contentAsText(), groupName);
228+
handleXdsResource(path, (JsonNode) change.content(), groupName);
227229
} catch (Throwable t) {
228230
logger.warn("Unexpected exception while handling an xDS resource from {}.",
229231
groupName + path, t);

0 commit comments

Comments
 (0)