Skip to content

Commit e89d1d4

Browse files
minwooxminwoox
authored
Add YAML/JSON backward compatibility for xDS resource files (#1324)
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.
1 parent eea921b commit e89d1d4

14 files changed

Lines changed: 974 additions & 50 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: 57 additions & 10 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.Consumer;
2627
import java.util.regex.Pattern;
2728

2829
import org.curioswitch.common.protobuf.json.MessageMarshaller;
@@ -179,12 +180,44 @@ private void checkWritePermission0(String group) {
179180
public <T extends Message> void push(
180181
StreamObserver<T> responseObserver, String group, String resourceName, String fileName,
181182
String summary, T resource, Author author, boolean create) {
183+
if (create) {
184+
// Before attempting to create, verify that neither the requested file nor its alternative
185+
// extension (.json ↔ .yaml) already exists, so a migrated .yaml resource is not silently
186+
// shadowed by a newly created .json file.
187+
final Repository repository = xdsProject.repos().get(group);
188+
final String altFileName = alternativeFileName(fileName);
189+
repository.find(Revision.HEAD, fileName + ',' + altFileName, FIND_ONE_WITHOUT_CONTENT)
190+
.handle((entries, cause) -> {
191+
if (cause != null) {
192+
responseObserver.onError(cause);
193+
return null;
194+
}
195+
if (!entries.isEmpty()) {
196+
responseObserver.onError(
197+
Status.ALREADY_EXISTS
198+
.withDescription("Resource already exists: " + resourceName)
199+
.asRuntimeException());
200+
return null;
201+
}
202+
doPush(responseObserver, group, resourceName, fileName, summary, resource, author, true);
203+
return null;
204+
});
205+
return;
206+
}
207+
doPush(responseObserver, group, resourceName, fileName, summary, resource, author, false);
208+
}
209+
210+
private <T extends Message> void doPush(
211+
StreamObserver<T> responseObserver, String group, String resourceName, String fileName,
212+
String summary, T resource, Author author, boolean create) {
182213
final Change<JsonNode> change;
183214
try {
184215
final String jsonText = JSON_MESSAGE_MARSHALLER.writeValueAsString(resource);
185216
final JsonNode jsonNode = Jackson.readTree(jsonText);
186217
if (create) {
187218
change = Change.ofJsonPatch(fileName, null, jsonNode);
219+
} else if (fileName.endsWith(".yaml")) {
220+
change = Change.ofYamlUpsert(fileName, jsonNode);
188221
} else {
189222
change = Change.ofJsonUpsert(fileName, jsonNode);
190223
}
@@ -236,9 +269,9 @@ public <T extends Message> void update(StreamObserver<T> responseObserver, Strin
236269
public <T extends Message> void update(StreamObserver<T> responseObserver, String group,
237270
String resourceName, String fileName, String summary, T resource,
238271
Author author) {
239-
updateOrDelete(responseObserver, group, resourceName, fileName,
240-
() -> push(responseObserver, group, resourceName, fileName,
241-
summary, resource, author, false));
272+
updateOrDelete(responseObserver, group, resourceName, fileName, resolvedFileName ->
273+
push(responseObserver, group, resourceName, resolvedFileName,
274+
summary, resource, author, false));
242275
}
243276

244277
public void delete(StreamObserver<Empty> responseObserver, String group,
@@ -248,10 +281,10 @@ public void delete(StreamObserver<Empty> responseObserver, String group,
248281

249282
public void delete(StreamObserver<Empty> responseObserver, String group,
250283
String resourceName, String fileName, String summary, Author author) {
251-
final Runnable deleteTask = () ->
284+
updateOrDelete(responseObserver, group, resourceName, fileName, resolvedFileName ->
252285
commandExecutor.execute(Command.push(author, XDS_CENTRAL_DOGMA_PROJECT, group,
253286
Revision.HEAD, summary, "", Markup.PLAINTEXT,
254-
ImmutableList.of(Change.ofRemoval(fileName))))
287+
ImmutableList.of(Change.ofRemoval(resolvedFileName))))
255288
.handle((unused, cause) -> {
256289
if (cause != null) {
257290
responseObserver.onError(
@@ -261,14 +294,17 @@ public void delete(StreamObserver<Empty> responseObserver, String group,
261294
responseObserver.onNext(Empty.getDefaultInstance());
262295
responseObserver.onCompleted();
263296
return null;
264-
});
265-
updateOrDelete(responseObserver, group, resourceName, fileName, deleteTask);
297+
}));
266298
}
267299

268300
public void updateOrDelete(StreamObserver<?> responseObserver, String group, String resourceName,
269-
String fileName, Runnable task) {
301+
String fileName, Consumer<String> taskProvider) {
270302
final Repository repository = xdsProject.repos().get(group);
271-
repository.find(Revision.HEAD, fileName, FIND_ONE_WITHOUT_CONTENT).handle((entries, cause) -> {
303+
// Search for both the requested filename and its alternative extension (.json ↔ .yaml)
304+
// to support files that may have been written in either format.
305+
final String altFileName = alternativeFileName(fileName);
306+
repository.find(Revision.HEAD, fileName + ',' + altFileName, FIND_ONE_WITHOUT_CONTENT)
307+
.handle((entries, cause) -> {
272308
if (cause != null) {
273309
responseObserver.onError(cause);
274310
return null;
@@ -279,8 +315,19 @@ public void updateOrDelete(StreamObserver<?> responseObserver, String group, Str
279315
.asRuntimeException());
280316
return null;
281317
}
282-
task.run();
318+
final String resolvedFileName = entries.keySet().iterator().next();
319+
taskProvider.accept(resolvedFileName);
283320
return null;
284321
});
285322
}
323+
324+
private static String alternativeFileName(String fileName) {
325+
if (fileName.endsWith(".json")) {
326+
return fileName.substring(0, fileName.length() - 5) + ".yaml";
327+
}
328+
if (fileName.endsWith(".yaml")) {
329+
return fileName.substring(0, fileName.length() - 5) + ".json";
330+
}
331+
return fileName;
332+
}
286333
}

0 commit comments

Comments
 (0)