Skip to content

Commit 1bcaf83

Browse files
jrhee17jrhee17
committed
minimal impl
1 parent 5ab0d65 commit 1bcaf83

13 files changed

Lines changed: 1511 additions & 6 deletions

File tree

‎client/java-armeria-xds/src/main/java/com/linecorp/centraldogma/client/armeria/xds/configsource/CachingStream.java‎

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
/*
2+
* Copyright 2026 LY Corporation
3+
*
4+
* LY Corporation licenses this file to you under the Apache License,
5+
* version 2.0 (the "License"); you may not use this file except in compliance
6+
* with the License. You may obtain a copy of the License at:
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12+
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13+
* License for the specific language governing permissions and limitations
14+
* under the License.
15+
*/
16+
package com.linecorp.centraldogma.client.armeria.xds.configsource;
17+
18+
import static java.util.Objects.requireNonNull;
19+
20+
import java.util.HashMap;
21+
import java.util.Map;
22+
import java.util.function.Function;
23+
24+
import com.linecorp.armeria.xds.SnapshotWatcher;
25+
import com.linecorp.armeria.xds.stream.RefCountedStream;
26+
import com.linecorp.armeria.xds.stream.SnapshotStream;
27+
import com.linecorp.armeria.xds.stream.Subscription;
28+
29+
// Forked from com.linecorp.armeria.xds.CachingStream
30+
final class CachingStream<K, T> {
31+
32+
private final Function<K, SnapshotStream<T>> factory;
33+
private final Map<K, CacheEntry> cache = new HashMap<>();
34+
35+
CachingStream(Function<K, SnapshotStream<T>> factory) {
36+
this.factory = requireNonNull(factory, "factory");
37+
}
38+
39+
SnapshotStream<T> subscribe(K key) {
40+
requireNonNull(key, "key");
41+
return watcher -> {
42+
final CacheEntry entry = cache.computeIfAbsent(key, k -> new CacheEntry(k));
43+
return entry.subscribe(watcher);
44+
};
45+
}
46+
47+
private final class CacheEntry extends RefCountedStream<T> {
48+
private final K key;
49+
50+
CacheEntry(K key) {
51+
this.key = key;
52+
}
53+
54+
@Override
55+
protected Subscription onStart(SnapshotWatcher<T> watcher) {
56+
return factory.apply(key).subscribe(watcher);
57+
}
58+
59+
@Override
60+
protected void onStop() {
61+
cache.remove(key);
62+
}
63+
}
64+
}

‎client/java-armeria-xds/src/main/java/com/linecorp/centraldogma/client/armeria/xds/configsource/CentralDogmaSotwConfigSourceSubscriptionFactory.java‎

Lines changed: 209 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,209 @@
1+
/*
2+
* Copyright 2026 LY Corporation
3+
*
4+
* LY Corporation licenses this file to you under the Apache License,
5+
* version 2.0 (the "License"); you may not use this file except in compliance
6+
* with the License. You may obtain a copy of the License at:
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12+
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13+
* License for the specific language governing permissions and limitations
14+
* under the License.
15+
*/
16+
package com.linecorp.centraldogma.client.armeria.xds.configsource;
17+
18+
import static com.google.common.base.Preconditions.checkArgument;
19+
20+
import java.util.ArrayList;
21+
import java.util.EnumMap;
22+
import java.util.List;
23+
import java.util.Map;
24+
25+
import com.fasterxml.jackson.databind.JsonNode;
26+
import com.fasterxml.jackson.databind.node.ObjectNode;
27+
import com.github.xds.centraldogma.v1.CentralDogmaConfigSource;
28+
import com.google.common.collect.ImmutableList;
29+
import com.google.common.collect.ImmutableMap;
30+
import com.google.protobuf.Any;
31+
32+
import com.linecorp.armeria.common.util.Exceptions;
33+
import com.linecorp.armeria.xds.ClusterSnapshot;
34+
import com.linecorp.armeria.xds.SnapshotWatcher;
35+
import com.linecorp.armeria.xds.XdsResourceReader;
36+
import com.linecorp.armeria.xds.XdsType;
37+
import com.linecorp.armeria.xds.configsource.InterestedResources;
38+
import com.linecorp.armeria.xds.configsource.SotwConfigSourceSubscriptionFactory;
39+
import com.linecorp.armeria.xds.filter.FactoryContext;
40+
import com.linecorp.armeria.xds.stream.RefCountedStream;
41+
import com.linecorp.armeria.xds.stream.SnapshotStream;
42+
import com.linecorp.armeria.xds.stream.Subscription;
43+
import com.linecorp.centraldogma.client.CentralDogma;
44+
import com.linecorp.centraldogma.client.Watcher;
45+
import com.linecorp.centraldogma.client.WatcherRequest;
46+
import com.linecorp.centraldogma.common.Query;
47+
import com.linecorp.centraldogma.internal.Jackson;
48+
49+
import io.envoyproxy.envoy.config.core.v3.ConfigSource;
50+
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
51+
import io.netty.util.concurrent.EventExecutor;
52+
53+
/**
54+
* A {@link SotwConfigSourceSubscriptionFactory} that fetches xDS resources from Central Dogma.
55+
*/
56+
public final class CentralDogmaSotwConfigSourceSubscriptionFactory
57+
implements SotwConfigSourceSubscriptionFactory {
58+
59+
static final String NAME = "centraldogma.config_source";
60+
static final String TYPE_URL =
61+
"type.googleapis.com/com.github.xds.centraldogma.v1.CentralDogmaConfigSource";
62+
63+
@Override
64+
public String name() {
65+
return NAME;
66+
}
67+
68+
@Override
69+
public List<String> typeUrls() {
70+
return ImmutableList.of(TYPE_URL);
71+
}
72+
73+
@Override
74+
public SnapshotStream<DiscoveryResponse> create(ConfigSource configSource,
75+
FactoryContext factoryContext,
76+
SnapshotStream<InterestedResources> interestedResources) {
77+
final CentralDogmaConfigSource cdConfig =
78+
factoryContext.validator().unpack(
79+
configSource.getCustomConfigSource(), CentralDogmaConfigSource.class);
80+
checkArgument(!cdConfig.getClusterName().isEmpty(),
81+
"CentralDogmaConfigSource.cluster_name must not be empty");
82+
final SnapshotStream<ClusterSnapshot> clusterStream =
83+
factoryContext.clusterStream(cdConfig.getClusterName());
84+
final EventExecutor eventLoop = factoryContext.eventLoop();
85+
final SnapshotStream<CentralDogma> cdStream =
86+
clusterStream.switchMapEager(CentralDogmaClientStream::new);
87+
return cdStream.switchMapEager(centralDogma -> {
88+
final Map<XdsType, InterestedResources> accumulated = new EnumMap<>(XdsType.class);
89+
final CachingStream<String, JsonNode> watcherCache = new CachingStream<>(
90+
name -> new WatcherStream(centralDogma, ResourcePath.parse(name))
91+
.rescheduleEventsOn(eventLoop));
92+
return interestedResources
93+
.map(interest -> {
94+
accumulated.put(interest.type(), interest);
95+
return ImmutableMap.copyOf(accumulated);
96+
})
97+
.switchMapEager(interests ->
98+
new Interest2ResponsesStream(interests, watcherCache));
99+
});
100+
}
101+
102+
private static final class Interest2ResponsesStream extends RefCountedStream<DiscoveryResponse> {
103+
104+
private final Map<XdsType, InterestedResources> interests;
105+
private final CachingStream<String, JsonNode> watcherCache;
106+
107+
Interest2ResponsesStream(Map<XdsType, InterestedResources> interests,
108+
CachingStream<String, JsonNode> watcherCache) {
109+
this.interests = interests;
110+
this.watcherCache = watcherCache;
111+
}
112+
113+
@Override
114+
protected Subscription onStart(SnapshotWatcher<DiscoveryResponse> watcher) {
115+
final List<Subscription> subs = new ArrayList<>();
116+
for (InterestedResources interested : interests.values()) {
117+
final String typeUrl = interested.type().typeUrl();
118+
final List<SnapshotStream<Any>> streams =
119+
interested.resourceNames().stream()
120+
.map(name -> watcherCache.subscribe(name)
121+
.map(jsonNode -> toAny(jsonNode, typeUrl)))
122+
.collect(ImmutableList.toImmutableList());
123+
subs.add(SnapshotStream.combineNLatest(streams)
124+
.map(resources -> DiscoveryResponse.newBuilder()
125+
.setTypeUrl(typeUrl)
126+
.addAllResources(resources)
127+
.build())
128+
.subscribe(this::emit));
129+
}
130+
return () -> {
131+
subs.forEach(Subscription::close);
132+
subs.clear();
133+
};
134+
}
135+
136+
private static Any toAny(JsonNode jsonNode, String typeUrl) {
137+
((ObjectNode) jsonNode).put("@type", typeUrl);
138+
return XdsResourceReader.from(jsonNode.toString(), Any.class);
139+
}
140+
}
141+
142+
private static final class CentralDogmaClientStream extends RefCountedStream<CentralDogma> {
143+
144+
private final ClusterSnapshot clusterSnapshot;
145+
146+
CentralDogmaClientStream(ClusterSnapshot clusterSnapshot) {
147+
this.clusterSnapshot = clusterSnapshot;
148+
}
149+
150+
@Override
151+
protected Subscription onStart(SnapshotWatcher<CentralDogma> watcher) {
152+
final CentralDogma centralDogma =
153+
PreprocessorCentralDogmaBuilder.of(clusterSnapshot.preprocessor());
154+
emit(centralDogma, null);
155+
return () -> {
156+
try {
157+
centralDogma.close();
158+
} catch (Exception e) {
159+
Exceptions.throwUnsafely(e);
160+
}
161+
};
162+
}
163+
}
164+
165+
private static final class WatcherStream extends RefCountedStream<JsonNode> {
166+
167+
private final CentralDogma centralDogma;
168+
private final ResourcePath resourcePath;
169+
170+
WatcherStream(CentralDogma centralDogma, ResourcePath resourcePath) {
171+
this.centralDogma = centralDogma;
172+
this.resourcePath = resourcePath;
173+
}
174+
175+
@Override
176+
protected Subscription onStart(SnapshotWatcher<JsonNode> watcher) {
177+
final Watcher<?> cdWatcher;
178+
if (resourcePath.isFtl()) {
179+
// .ftl files are stored as TEXT, so we use Query.ofText and parse after rendering.
180+
final WatcherRequest<String> textRequest =
181+
centralDogma.forRepo(resourcePath.project(), resourcePath.repo())
182+
.watcher(Query.ofText(resourcePath.path()));
183+
textRequest.renderTemplate(true);
184+
if (resourcePath.profile() != null) {
185+
textRequest.renderTemplate(resourcePath.profile());
186+
}
187+
final Watcher<String> textWatcher = textRequest.start();
188+
textWatcher.watch((revision, text) -> emitText(text));
189+
cdWatcher = textWatcher;
190+
} else {
191+
final WatcherRequest<JsonNode> jsonRequest =
192+
centralDogma.forRepo(resourcePath.project(), resourcePath.repo())
193+
.watcher(resourcePath.query());
194+
final Watcher<JsonNode> jsonWatcher = jsonRequest.start();
195+
jsonWatcher.watch((revision, jsonNode) -> emit(jsonNode, null));
196+
cdWatcher = jsonWatcher;
197+
}
198+
return cdWatcher::close;
199+
}
200+
201+
private void emitText(String text) {
202+
try {
203+
emit(Jackson.readTree(resourcePath.basePath(), text), null);
204+
} catch (Exception e) {
205+
emit(null, e);
206+
}
207+
}
208+
}
209+
}

‎client/java-armeria-xds/src/main/java/com/linecorp/centraldogma/client/armeria/xds/configsource/PreprocessorCentralDogmaBuilder.java‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
1+
/*
2+
* Copyright 2026 LY Corporation
3+
*
4+
* LY Corporation licenses this file to you under the Apache License,
5+
* version 2.0 (the "License"); you may not use this file except in compliance
6+
* with the License. You may obtain a copy of the License at:
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12+
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13+
* License for the specific language governing permissions and limitations
14+
* under the License.
15+
*/
16+
package com.linecorp.centraldogma.client.armeria.xds.configsource;
17+
18+
import static java.util.Objects.requireNonNull;
19+
20+
import com.linecorp.armeria.client.ClientBuilder;
21+
import com.linecorp.armeria.client.ClientPreprocessors;
22+
import com.linecorp.armeria.client.Clients;
23+
import com.linecorp.armeria.client.HttpPreprocessor;
24+
import com.linecorp.armeria.client.WebClient;
25+
import com.linecorp.armeria.client.encoding.DecodingClient;
26+
import com.linecorp.armeria.common.CommonPools;
27+
import com.linecorp.centraldogma.client.CentralDogma;
28+
import com.linecorp.centraldogma.internal.client.armeria.ArmeriaCentralDogma;
29+
30+
/**
31+
* Creates a {@link CentralDogma} client that connects through an {@link HttpPreprocessor}.
32+
*/
33+
final class PreprocessorCentralDogmaBuilder {
34+
35+
static CentralDogma of(HttpPreprocessor preprocessor) {
36+
requireNonNull(preprocessor, "preprocessor");
37+
final ClientBuilder builder =
38+
Clients.builder(ClientPreprocessors.of(preprocessor));
39+
builder.decorator(DecodingClient.newDecorator());
40+
final WebClient client = builder.build(WebClient.class);
41+
return new ArmeriaCentralDogma(CommonPools.blockingTaskExecutor(), client, "anonymous",
42+
() -> {}, null, null);
43+
}
44+
45+
private PreprocessorCentralDogmaBuilder() {}
46+
}

0 commit comments

Comments
 (0)