New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Add CentralDogma backed EndpointGroup #105
Changes from 5 commits
dc4157d
b063815
ec755f4
5eff202
3390f9c
4dc4e5d
abfb110
5fabf32
91a6fd6
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,129 @@ | ||
/* | ||
* Copyright 2018 LINE Corporation | ||
* | ||
* LINE 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.client.armeria; | ||
|
||
import static java.util.Objects.requireNonNull; | ||
|
||
import java.util.List; | ||
import java.util.concurrent.TimeUnit; | ||
import java.util.concurrent.TimeoutException; | ||
|
||
import org.slf4j.Logger; | ||
import org.slf4j.LoggerFactory; | ||
|
||
import com.fasterxml.jackson.databind.JsonNode; | ||
|
||
import com.linecorp.armeria.client.Endpoint; | ||
import com.linecorp.armeria.client.endpoint.DynamicEndpointGroup; | ||
import com.linecorp.armeria.client.endpoint.EndpointGroup; | ||
import com.linecorp.centraldogma.client.CentralDogma; | ||
import com.linecorp.centraldogma.client.Watcher; | ||
import com.linecorp.centraldogma.common.Query; | ||
|
||
/** | ||
* A CentralDogma based {@link EndpointGroup} implementation. This {@link EndpointGroup} retrieves the list of | ||
* {@link Endpoint}s from a route file served by CentralDogma, and update the list when upstream data changes. | ||
* Route file could be JSON file or normal text file. | ||
* | ||
* <p>For example, the following JSON file will be served as a route file: | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
* <pre>{@code | ||
* [ | ||
* "host1:port1", | ||
* "host2:port2", | ||
* "host3:port3" | ||
* ] | ||
* }</pre> | ||
* | ||
* <p>The route file could be retrieved as an {@link EndpointGroup} using the following code: | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The JSON array file could be.. |
||
* <pre>{@code | ||
* CentralDogmaEndpointGroup<JsonNode> endpointGroup = CentralDogmaEndpointGroup.of( | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Could be indented one less column? |
||
* centralDogma, "myProject", "myRepo", | ||
* Query.ofJsonPath("/route.json"), | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
* EndpointListDecoder.JSON | ||
* ) | ||
* endpointGroup.endpoints(); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Don't we need to wait for the initial retrieval? |
||
* }</pre> | ||
* | ||
* @param <T> Type of CentralDomgma file (could be {@link JsonNode} or {@link String}) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. the type of the file in Central Dogma? |
||
*/ | ||
public final class CentralDogmaEndpointGroup<T> extends DynamicEndpointGroup { | ||
private static final Logger logger = LoggerFactory.getLogger(CentralDogmaEndpointGroup.class); | ||
private static final long WATCH_INITIALIZATION_TIMEOUT_MILLIS = TimeUnit.SECONDS.toMillis(10); | ||
|
||
private final Watcher<T> instanceListWatcher; | ||
private final EndpointListDecoder<T> endpointCodec; | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I'm not sure if we need a different name for this. Codec means encoder and decoder, but this seems like only decoding now. |
||
|
||
/** | ||
* Creates a new {@link CentralDogmaEndpointGroup}. | ||
* | ||
* @param watcher A {@link Watcher} | ||
* @param endpointCodec A {@link EndpointListDecoder} | ||
*/ | ||
public static <T> CentralDogmaEndpointGroup<T> ofWatcher(Watcher<T> watcher, | ||
EndpointListDecoder<T> endpointCodec) { | ||
return new CentralDogmaEndpointGroup<>(watcher, endpointCodec); | ||
} | ||
|
||
/** | ||
* Creates a new {@link CentralDogmaEndpointGroup}. | ||
* | ||
* @param centralDogma A {@link CentralDogma} | ||
* @param projectName CentralDogma project name | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. global: CentralDogma -> Central Dogma |
||
* @param repositoryName CentralDogma repository name | ||
* @param query A {@link Query} to route file | ||
* @param endpointCodec An {@link EndpointListDecoder} | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. A -> a, An -> an |
||
*/ | ||
public static <T> CentralDogmaEndpointGroup<T> of(CentralDogma centralDogma, | ||
String projectName, String repositoryName, | ||
Query<T> query, | ||
EndpointListDecoder<T> endpointCodec) { | ||
return ofWatcher(centralDogma.fileWatcher(projectName, repositoryName, query), endpointCodec); | ||
} | ||
|
||
private CentralDogmaEndpointGroup(Watcher<T> instanceListWatcher, EndpointListDecoder<T> endpointCodec) { | ||
this.instanceListWatcher = requireNonNull(instanceListWatcher, "instanceListWatcher"); | ||
this.endpointCodec = requireNonNull(endpointCodec, "endpointCodec"); | ||
registerWatcher(); | ||
} | ||
|
||
private void registerWatcher() { | ||
instanceListWatcher.watch((revision, instances) -> { | ||
try { | ||
List<Endpoint> newEndpoints = endpointCodec.decode(instances); | ||
if (newEndpoints.isEmpty()) { | ||
logger.info("Not refreshing the endpoint list of {} because it's empty. {}", | ||
instanceListWatcher, revision); | ||
return; | ||
} | ||
setEndpoints(newEndpoints); | ||
} catch (Exception e) { | ||
logger.warn("Failed to refresh the endpoint list.", e); | ||
} | ||
}); | ||
try { | ||
instanceListWatcher.awaitInitialValue(WATCH_INITIALIZATION_TIMEOUT_MILLIS, | ||
TimeUnit.MILLISECONDS); | ||
} catch (InterruptedException | TimeoutException e) { | ||
logger.warn("Failed to initialize instance list on time.", e); | ||
} | ||
} | ||
|
||
@Override | ||
public void close() { | ||
instanceListWatcher.close(); | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,39 @@ | ||
/* | ||
* Copyright 2018 LINE Corporation | ||
* | ||
* LINE 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.client.armeria; | ||
|
||
import static com.fasterxml.jackson.databind.DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES; | ||
import static com.google.common.collect.ImmutableList.toImmutableList; | ||
|
||
import java.util.List; | ||
|
||
import com.fasterxml.jackson.databind.ObjectMapper; | ||
|
||
import com.linecorp.armeria.client.Endpoint; | ||
|
||
final class EndpointListCodecUtils { | ||
static final ObjectMapper objectMapper = new ObjectMapper().disable(FAIL_ON_UNKNOWN_PROPERTIES); | ||
|
||
private EndpointListCodecUtils() {} | ||
|
||
static List<Endpoint> convertToEndpointList(List<String> endpoints) { | ||
return endpoints.stream() | ||
.map(String::trim) | ||
.map(Endpoint::parse) | ||
.collect(toImmutableList()); | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,70 @@ | ||
/* | ||
* Copyright 2017 LINE Corporation | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 2018? |
||
* | ||
* LINE 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.client.armeria; | ||
|
||
import java.util.List; | ||
|
||
import com.fasterxml.jackson.databind.JsonNode; | ||
|
||
import com.linecorp.armeria.client.Endpoint; | ||
|
||
/** | ||
* Decode and encode between CentralDogma upstream file content and list of {@link Endpoint}s. | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
* | ||
* @param <T> Type of CentralDogma file (could be {@link JsonNode} or {@link String}) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. the type of the file in Central Dogma? |
||
*/ | ||
@FunctionalInterface | ||
public interface EndpointListDecoder<T> { | ||
|
||
/** | ||
* Default {@link EndpointListDecoder} implementation for JsonNode object. | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
* Retrive object must be a json array (which has format as "[ \"segment1\", \"segment2\" ]" | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
* Each segment represents an endpoint whose format is | ||
* {@code <host>[:<port_number>[:weight]]}, such as: | ||
* <ul> | ||
* <li>{@code "foo.com"} - default port number, default weight (1000)</li> | ||
* <li>{@code "bar.com:8080} - port number 8080, default weight (1000)</li> | ||
* <li>{@code "10.0.2.15:0:500} - default port number, weight 500</li> | ||
* <li>{@code "192.168.1.2:8443:700} - port number 8443, weight 700</li> | ||
* </ul> | ||
* Note that the port number must be specified when you want to specify the weight. | ||
*/ | ||
EndpointListDecoder<JsonNode> JSON = new JsonEndpointListDecoder(); | ||
|
||
/** | ||
* Default {@link EndpointListDecoder} implementation for String object. | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. String object -> |
||
* Retrive object must be a string which is a list of segments separated by newline. | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
* Each segment represents an endpoint whose format is | ||
* {@code <host>[:<port_number>[:weight]]}, such as: | ||
* <ul> | ||
* <li>{@code "foo.com"} - default port number, default weight (1000)</li> | ||
* <li>{@code "bar.com:8080} - port number 8080, default weight (1000)</li> | ||
* <li>{@code "10.0.2.15:0:500} - default port number, weight 500</li> | ||
* <li>{@code "192.168.1.2:8443:700} - port number 8443, weight 700</li> | ||
* </ul> | ||
* Note that the port number must be specified when you want to specify the weight. | ||
*/ | ||
EndpointListDecoder<String> TEXT = new TextEndpointListDecoder(); | ||
|
||
/** | ||
* Decodes an object into a set of {@link Endpoint}s. | ||
* | ||
* @param object An object retrieve from CentralDogma (could be JsonNode or String) | ||
* | ||
* @return the list of {@link Endpoint}s | ||
*/ | ||
List<Endpoint> decode(T object); | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,41 @@ | ||
/* | ||
* Copyright 2018 LINE Corporation | ||
* | ||
* LINE 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.client.armeria; | ||
|
||
import static com.linecorp.centraldogma.client.armeria.EndpointListCodecUtils.convertToEndpointList; | ||
import static com.linecorp.centraldogma.client.armeria.EndpointListCodecUtils.objectMapper; | ||
|
||
import java.io.IOException; | ||
import java.util.List; | ||
|
||
import com.fasterxml.jackson.core.type.TypeReference; | ||
import com.fasterxml.jackson.databind.JsonNode; | ||
|
||
import com.linecorp.armeria.client.Endpoint; | ||
|
||
final class JsonEndpointListDecoder implements EndpointListDecoder<JsonNode> { | ||
@Override | ||
public List<Endpoint> decode(JsonNode node) { | ||
final List<String> endpoints; | ||
try { | ||
endpoints = objectMapper.readValue(node.traverse(), | ||
new TypeReference<List<String>>() {}); | ||
} catch (IOException e) { | ||
throw new IllegalArgumentException("invalid format: " + node); | ||
} | ||
return convertToEndpointList(endpoints); | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,36 @@ | ||
/* | ||
* Copyright 2018 LINE Corporation | ||
* | ||
* LINE 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.client.armeria; | ||
|
||
import static com.linecorp.centraldogma.client.armeria.EndpointListCodecUtils.convertToEndpointList; | ||
|
||
import java.util.List; | ||
|
||
import com.google.common.base.CharMatcher; | ||
import com.google.common.base.Splitter; | ||
|
||
import com.linecorp.armeria.client.Endpoint; | ||
|
||
final class TextEndpointListDecoder implements EndpointListDecoder<String> { | ||
private static final Splitter NEWLINE_SPLITTER = Splitter.on(CharMatcher.anyOf("\n\r")) | ||
.omitEmptyStrings() | ||
.trimResults(); | ||
|
||
@Override | ||
public List<Endpoint> decode(String object) { | ||
return convertToEndpointList(NEWLINE_SPLITTER.splitToList(object)); | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,57 @@ | ||
/* | ||
* Copyright 2018 LINE Corporation | ||
* | ||
* LINE 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.client.armeria; | ||
|
||
import static org.assertj.core.api.Assertions.assertThat; | ||
|
||
import java.util.List; | ||
|
||
import org.junit.Test; | ||
|
||
import com.fasterxml.jackson.databind.JsonNode; | ||
import com.fasterxml.jackson.databind.ObjectMapper; | ||
import com.google.common.collect.ImmutableList; | ||
|
||
import com.linecorp.armeria.client.Endpoint; | ||
|
||
public class JsonEndpointListDecoderTest { | ||
private static final ObjectMapper objectMapper = new ObjectMapper(); | ||
|
||
static List<String> HOST_AND_PORT_LIST = ImmutableList.of( | ||
"centraldogma-sample001.com", | ||
"centraldogma-sample001.com:1234", | ||
"1.2.3.4", | ||
"1.2.3.4:5678" | ||
); | ||
|
||
static List<Endpoint> ENDPOINT_LIST = ImmutableList.of( | ||
Endpoint.of("centraldogma-sample001.com"), | ||
Endpoint.of("centraldogma-sample001.com", 1234), | ||
Endpoint.of("1.2.3.4"), | ||
Endpoint.of("1.2.3.4", 5678) | ||
); | ||
|
||
@Test | ||
public void decode() throws Exception { | ||
EndpointListDecoder<JsonNode> decoder = EndpointListDecoder.JSON; | ||
List<Endpoint> decoded = decoder.decode( | ||
objectMapper.readTree(objectMapper.writeValueAsString(HOST_AND_PORT_LIST))); | ||
|
||
assertThat(decoded).hasSize(4); | ||
assertThat(decoded).isEqualTo(ENDPOINT_LIST); | ||
} | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
A {@link DynamicEndpointGroup} implementation that retrieves the {@link Endpoint} list from an entry in Central Dogma. The entry can be a JSON file or a plain text file.