Skip to content
This repository was archived by the owner on Aug 25, 2024. It is now read-only.

Commit 146df93

Browse files
nicoloboschinicoloboschi
authored
Base support for templating (#16)
1 parent be541f1 commit 146df93

File tree

10 files changed

+399
-12
lines changed

10 files changed

+399
-12
lines changed

‎api/src/main/java/com/datastax/oss/sga/api/model/StreamingCluster.java‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -15,9 +15,6 @@
1515
*/
1616
package com.datastax.oss.sga.api.model;
1717

18-
import lombok.Builder;
19-
import lombok.Data;
20-
2118
import java.util.HashMap;
2219
import java.util.Map;
2320

‎api/src/main/java/com/datastax/oss/sga/api/model/TopicDefinition.java‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -27,24 +27,26 @@ public class TopicDefinition extends Connection.Connectable {
2727
public static final String CREATE_MODE_CREATE_IF_NOT_EXISTS = "create-if-not-exists";
2828

2929
public TopicDefinition() {
30-
if (creationMode == null) {
31-
creationMode = CREATE_MODE_NONE;
32-
}
30+
creationMode = CREATE_MODE_NONE;
3331
connectableType = Connection.Connectables.TOPIC;
3432
}
3533

3634
public TopicDefinition(String name, String creationMode, SchemaDefinition schema) {
3735
this();
3836
this.name = name;
39-
this.creationMode = creationMode;
37+
if (creationMode == null) {
38+
this.creationMode = CREATE_MODE_NONE;
39+
} else {
40+
this.creationMode = creationMode;
41+
}
4042
this.schema = schema;
4143
validateCreationMode();
4244
}
4345

4446
private String name;
4547

4648
@JsonProperty("creation-mode")
47-
private String creationMode = CREATE_MODE_NONE;
49+
private String creationMode;
4850
private SchemaDefinition schema;
4951

5052
private void validateCreationMode() {

‎api/src/main/java/com/datastax/oss/sga/api/runtime/ClusterRuntimeRegistry.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@
1313
*/
1414
public class ClusterRuntimeRegistry {
1515

16-
private final Map<String, ClusterRuntime<?>> registry = new ConcurrentHashMap<>();
16+
protected final Map<String, ClusterRuntime<?>> registry = new ConcurrentHashMap<>();
1717

1818
public ClusterRuntime getClusterRuntime(StreamingCluster streamingCluster) {
1919
Objects.requireNonNull(streamingCluster);

‎core/pom.xml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,5 +59,9 @@
5959
<artifactId>junit-jupiter</artifactId>
6060
<scope>test</scope>
6161
</dependency>
62+
<dependency>
63+
<groupId>com.samskivert</groupId>
64+
<artifactId>jmustache</artifactId>
65+
</dependency>
6266
</dependencies>
6367
</project>

‎core/src/main/java/com/datastax/oss/sga/impl/common/ApplicationInstancePlaceholderResolver.java‎

Lines changed: 136 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,136 @@
1+
package com.datastax.oss.sga.impl.common;
2+
3+
import com.datastax.oss.sga.api.model.AgentConfiguration;
4+
import com.datastax.oss.sga.api.model.ApplicationInstance;
5+
import com.datastax.oss.sga.api.model.Instance;
6+
import com.datastax.oss.sga.api.model.Module;
7+
import com.datastax.oss.sga.api.model.Pipeline;
8+
import com.datastax.oss.sga.api.model.Resource;
9+
import com.datastax.oss.sga.api.model.StreamingCluster;
10+
import com.fasterxml.jackson.databind.ObjectMapper;
11+
import com.samskivert.mustache.Mustache;
12+
import java.io.IOException;
13+
import java.util.HashMap;
14+
import java.util.LinkedHashMap;
15+
import java.util.Map;
16+
import lombok.SneakyThrows;
17+
18+
public class ApplicationInstancePlaceholderResolver {
19+
20+
private static final ObjectMapper mapper = new ObjectMapper();
21+
22+
private ApplicationInstancePlaceholderResolver() {
23+
}
24+
25+
@SneakyThrows
26+
public static ApplicationInstance resolvePlaceholders(ApplicationInstance instance) {
27+
instance = deepCopy(instance);
28+
final Map<String, Object> context = createContext(instance);
29+
30+
31+
instance.setInstance(resolveInstance(instance, context));
32+
instance.setResources(resolveResources(instance, context));
33+
instance.setModules(resolveModules(instance, context));
34+
return instance;
35+
}
36+
37+
static Map<String, Object> createContext(ApplicationInstance application) throws IOException {
38+
Map<String, Object> context = new HashMap<>();
39+
final Instance instance = application.getInstance();
40+
if (instance != null) {
41+
context.put("cluster", instance.streamingCluster());
42+
context.put("globals", instance.globals());
43+
}
44+
45+
Map<String, Map<String, Object>> secrets = new HashMap<>();
46+
if (application.getSecrets() != null && application.getSecrets().secrets() != null) {
47+
application.getSecrets().secrets().forEach((k, v) -> secrets.put(k, v.data()));
48+
}
49+
context.put("secrets", secrets);
50+
context = deepCopy(context);
51+
return context;
52+
}
53+
54+
private static Map<String, Module> resolveModules(ApplicationInstance instance, Map<String, Object> context) {
55+
Map<String, Module> newModules = new LinkedHashMap<>();
56+
for (Map.Entry<String, Module> moduleEntry : instance.getModules().entrySet()) {
57+
final Module module = moduleEntry.getValue();
58+
for (Map.Entry<String, Pipeline> pipelineEntry : module.getPipelines().entrySet()) {
59+
final Pipeline pipeline = pipelineEntry.getValue();
60+
Map<String, AgentConfiguration> newAgents = new LinkedHashMap<>();
61+
for (Map.Entry<String, AgentConfiguration> stringAgentConfigurationEntry : pipeline.getAgents()
62+
.entrySet()) {
63+
final AgentConfiguration value = stringAgentConfigurationEntry.getValue();
64+
value.setConfiguration(resolveMap(context, value.getConfiguration()));
65+
newAgents.put(stringAgentConfigurationEntry.getKey(), value);
66+
}
67+
pipeline.setAgents(newAgents);
68+
}
69+
newModules.put(moduleEntry.getKey(), module);
70+
}
71+
return newModules;
72+
}
73+
74+
private static Instance resolveInstance(ApplicationInstance applicationInstance, Map<String, Object> context) {
75+
final StreamingCluster newCluster;
76+
final Instance instance = applicationInstance.getInstance();
77+
if (instance == null) {
78+
return null;
79+
}
80+
final StreamingCluster cluster = instance.streamingCluster();
81+
if (cluster != null) {
82+
newCluster = new StreamingCluster(cluster.type(), resolveMap(context, cluster.configuration()));
83+
} else {
84+
newCluster = null;
85+
}
86+
return new Instance(
87+
newCluster,
88+
resolveMap(context, instance.globals())
89+
);
90+
}
91+
92+
private static Map<String, Resource> resolveResources(ApplicationInstance instance,
93+
Map<String, Object> context) {
94+
Map<String, Resource> newResources = new HashMap<>();
95+
for (Map.Entry<String, Resource> resourceEntry : instance.getResources().entrySet()) {
96+
final Resource resource = resourceEntry.getValue();
97+
newResources.put(resourceEntry.getKey(),
98+
new Resource(
99+
resource.id(),
100+
resource.name(),
101+
resource.type(),
102+
resolveMap(context, resource.configuration())
103+
)
104+
);
105+
}
106+
return newResources;
107+
}
108+
109+
static Map<String, Object> resolveMap(Map<String, Object> context, Map<String, Object> config) {
110+
111+
Map<String, Object> resolvedConfig = new HashMap<>();
112+
if (config == null) {
113+
return resolvedConfig;
114+
}
115+
for (Map.Entry<String, Object> stringObjectEntry : config.entrySet()) {
116+
resolvedConfig.put(stringObjectEntry.getKey(), resolveValue(context, stringObjectEntry.getValue() + ""));
117+
}
118+
return resolvedConfig;
119+
}
120+
121+
static String resolveValue(Map<String, Object> context, String template) {
122+
return Mustache.compiler()
123+
.compile(template)
124+
.execute(context);
125+
}
126+
127+
private static ApplicationInstance deepCopy(ApplicationInstance instance) throws IOException {
128+
return mapper.readValue(mapper.writeValueAsBytes(instance), ApplicationInstance.class);
129+
}
130+
131+
private static Map<String, Object> deepCopy(Map<String, Object> context) throws IOException {
132+
return mapper.readValue(mapper.writeValueAsBytes(context), Map.class);
133+
}
134+
135+
136+
}

‎core/src/main/java/com/datastax/oss/sga/impl/common/PlaceholderEvaluator.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
package com.datastax.oss.sga.impl.common;
2+
3+
import com.datastax.oss.sga.api.model.ApplicationInstance;
4+
5+
public class PlaceholderEvaluator {
6+
7+
public static ApplicationInstance evaluate(ApplicationInstance applicationInstance) {
8+
new ApplicationInstance();
9+
return applicationInstance;
10+
}
11+
12+
}

‎core/src/main/java/com/datastax/oss/sga/impl/deploy/ApplicationDeployer.java‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -5,13 +5,14 @@
55
import com.datastax.oss.sga.api.runtime.PhysicalApplicationInstance;
66
import com.datastax.oss.sga.api.runtime.PluginsRegistry;
77
import com.datastax.oss.sga.api.runtime.ClusterRuntimeRegistry;
8+
import com.datastax.oss.sga.impl.common.ApplicationInstancePlaceholderResolver;
89
import lombok.Builder;
910

1011
@Builder
1112
public class ApplicationDeployer<T extends PhysicalApplicationInstance> {
1213

13-
private ClusterRuntimeRegistry registry = new ClusterRuntimeRegistry();
14-
private PluginsRegistry pluginsRegistry = new PluginsRegistry();
14+
private ClusterRuntimeRegistry registry;
15+
private PluginsRegistry pluginsRegistry;
1516

1617
public T createImplementation(ApplicationInstance applicationInstance) {
1718
ClusterRuntime<T> clusterRuntime = registry.getClusterRuntime(applicationInstance.getInstance().streamingCluster());
@@ -20,7 +21,9 @@ public T createImplementation(ApplicationInstance applicationInstance) {
2021

2122
public void deploy(ApplicationInstance applicationInstance, T physicalApplicationInstance) {
2223
ClusterRuntime<T> clusterRuntime = registry.getClusterRuntime(applicationInstance.getInstance().streamingCluster());
23-
clusterRuntime.deploy(applicationInstance, physicalApplicationInstance);
24+
final ApplicationInstance resolvedApplicationInstance = ApplicationInstancePlaceholderResolver
25+
.resolvePlaceholders(applicationInstance);
26+
clusterRuntime.deploy(resolvedApplicationInstance, physicalApplicationInstance);
2427
}
2528

2629
public void delete(ApplicationInstance applicationInstance, T physicalApplicationInstance) {

‎core/src/test/java/com/datastax/oss/sga/impl/common/ApplicationInstancePlaceholderResolverTest.java‎

Lines changed: 135 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,135 @@
1+
package com.datastax.oss.sga.impl.common;
2+
3+
import com.datastax.oss.sga.api.model.ApplicationInstance;
4+
import com.datastax.oss.sga.api.model.Resource;
5+
import com.datastax.oss.sga.impl.parser.ModelBuilder;
6+
import com.samskivert.mustache.MustacheException;
7+
import java.util.Map;
8+
import org.junit.jupiter.api.Assertions;
9+
import org.junit.jupiter.api.Test;
10+
11+
class ApplicationInstancePlaceholderResolverTest {
12+
13+
@Test
14+
void testAvailablePlaceholders() throws Exception {
15+
16+
ApplicationInstance applicationInstance = ModelBuilder
17+
.buildApplicationInstance(Map.of(
18+
"secrets.yaml", """
19+
secrets:
20+
- name: "OpenAI Azure credentials"
21+
id: "openai-credentials"
22+
data:
23+
accessKey: "my-access-key"
24+
""",
25+
"instance.yaml", """
26+
instance:
27+
streamingCluster:
28+
type: pulsar
29+
configuration:
30+
webServiceUrl: http://mypulsar.localhost:8080
31+
globals:
32+
another-url: another-value
33+
open-api-url: http://myurl.localhost:8080/endpoint
34+
"""));
35+
36+
final Map<String, Object> context = ApplicationInstancePlaceholderResolver.createContext(applicationInstance);
37+
Assertions.assertEquals("my-access-key",
38+
ApplicationInstancePlaceholderResolver.resolveValue(context,
39+
"{{secrets.openai-credentials.accessKey}}"));
40+
Assertions.assertEquals("http://mypulsar.localhost:8080",
41+
ApplicationInstancePlaceholderResolver.resolveValue(context,
42+
"{{cluster.configuration.webServiceUrl}}"));
43+
Assertions.assertEquals("http://myurl.localhost:8080/endpoint",
44+
ApplicationInstancePlaceholderResolver.resolveValue(context, "{{globals.open-api-url}}"));
45+
}
46+
47+
@Test
48+
void testResolveSecretsInConfiguration() throws Exception {
49+
ApplicationInstance applicationInstance = ModelBuilder
50+
.buildApplicationInstance(Map.of("configuration.yaml",
51+
"""
52+
configuration:
53+
resources:
54+
- type: "openai-azure-config"
55+
name: "OpenAI Azure configuration"
56+
id: "openai-azure"
57+
configuration:
58+
credentials: "{{secrets.openai-credentials.accessKey}}"
59+
url: "{{globals.open-api-url}}"
60+
61+
""",
62+
"secrets.yaml", """
63+
secrets:
64+
- name: "OpenAI Azure credentials"
65+
id: "openai-credentials"
66+
data:
67+
accessKey: "my-access-key"
68+
""",
69+
"instance.yaml", """
70+
instance:
71+
globals:
72+
another-url: another-value
73+
open-api-url: http://myurl.localhost:8080/endpoint
74+
"""));
75+
76+
final ApplicationInstance resolved =
77+
ApplicationInstancePlaceholderResolver.resolvePlaceholders(applicationInstance);
78+
final Resource resource = resolved.getResources().get("openai-azure");
79+
Assertions.assertEquals("my-access-key", resource.configuration().get("credentials"));
80+
Assertions.assertEquals("http://myurl.localhost:8080/endpoint", resource.configuration().get("url"));
81+
}
82+
83+
@Test
84+
void testResolveInAgentConfiguration() throws Exception {
85+
ApplicationInstance applicationInstance = ModelBuilder
86+
.buildApplicationInstance(Map.of("module1.yaml",
87+
"""
88+
module: "module-1"
89+
id: "pipeline-1"
90+
topics:
91+
- name: "input-topic"
92+
pipeline:
93+
- name: "sink1"
94+
id: "sink1"
95+
type: "generic-pulsar-sink"
96+
input: "input-topic"
97+
configuration:
98+
sinkType: "some-sink-type-on-your-cluster"
99+
access-key: "{{ secrets.ak.value }}"
100+
""",
101+
"secrets.yaml", """
102+
secrets:
103+
- name: "OpenAI Azure credentials"
104+
id: "ak"
105+
data:
106+
value: "my-access-key"
107+
"""));
108+
109+
final ApplicationInstance resolved =
110+
ApplicationInstancePlaceholderResolver.resolvePlaceholders(applicationInstance);
111+
Assertions.assertEquals("my-access-key",
112+
resolved.getModule("module-1").getPipelines().values().iterator().next().getAgents().get("sink1")
113+
.getConfiguration()
114+
.get("access-key"));
115+
}
116+
117+
@Test
118+
void testErrorOnNotFound() throws Exception {
119+
ApplicationInstance applicationInstance = ModelBuilder
120+
.buildApplicationInstance(Map.of("configuration.yaml",
121+
"""
122+
configuration:
123+
resources:
124+
- type: "openai-azure-config"
125+
name: "OpenAI Azure configuration"
126+
id: "openai-azure"
127+
configuration:
128+
credentials: "{{secrets.openai-credentials.invalid}}"
129+
130+
"""));
131+
Assertions.assertThrows(MustacheException.Context.class, () -> {
132+
ApplicationInstancePlaceholderResolver.resolvePlaceholders(applicationInstance);
133+
});
134+
}
135+
}

0 commit comments

Comments
 (0)