This is an automated email from the ASF dual-hosted git repository.
marat pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-karavan.git
commit b383fbc04ba7fb51ca53e624e1af438247c284ae
Author: Marat Gubaidullin <marat.gubaidul...@gmail.com>
AuthorDate: Fri Jul 7 19:18:49 2023 -0400
Runner experiment #817
---
.../apache/camel/karavan/api/RunnerResource.java | 13 +-
.../karavan/listener/ClientRunnerListener.java | 46 +++++
.../karavan/listener/LocalRunnerListener.java | 48 +++++
.../camel/karavan/listener/RunnerListener.java | 35 ++++
.../org/apache/camel/karavan/model/GroupedKey.java | 12 ++
.../apache/camel/karavan/model/RunnerCommand.java | 13 ++
.../camel/karavan/service/InfinispanService.java | 105 ++++++-----
.../camel/karavan/service/KubernetesService.java | 64 ++++---
.../camel/karavan/service/RunnerService.java | 22 ++-
.../src/main/resources/application.properties | 2 -
karavan-bashi/pom.xml | 4 +
.../{KaravanBashi.java => ConductorService.java} | 50 +++--
.../{KaravanConstants.java => Constants.java} | 5 +-
.../apache/camel/karavan/bashi/HealthChecker.java | 31 ---
.../apache/camel/karavan/bashi/KaravanBashi.java | 65 +------
.../camel/karavan/bashi/KaravanContainers.java | 33 ----
.../camel/karavan/bashi/RunnerStatusService.java | 83 ++++++++
.../karavan/bashi/docker/DockerEventListener.java | 47 +++--
.../camel/karavan/bashi/docker/DockerService.java | 74 ++++++--
.../bashi/infinispan/ClientRunnerListener.java | 49 +++++
.../karavan/bashi/infinispan}/GroupedKey.java | 14 +-
.../bashi/infinispan/InfinispanService.java | 122 ++++++++++++
.../camel/karavan/bashi/infinispan/PodStatus.java | 209 +++++++++++++++++++++
.../bashi/infinispan/ProjectStoreSchema.java | 8 +
.../karavan/bashi/infinispan/RunnerCommand.java | 13 ++
.../src/main/resources/application.properties | 10 +
26 files changed, 902 insertions(+), 275 deletions(-)
diff --git
a/karavan-app/src/main/java/org/apache/camel/karavan/api/RunnerResource.java
b/karavan-app/src/main/java/org/apache/camel/karavan/api/RunnerResource.java
index b728f7f4..712be7d1 100644
--- a/karavan-app/src/main/java/org/apache/camel/karavan/api/RunnerResource.java
+++ b/karavan-app/src/main/java/org/apache/camel/karavan/api/RunnerResource.java
@@ -18,9 +18,9 @@ package org.apache.camel.karavan.api;
import org.apache.camel.karavan.model.PodStatus;
import org.apache.camel.karavan.model.Project;
+import org.apache.camel.karavan.model.RunnerCommand;
import org.apache.camel.karavan.model.RunnerStatus;
import org.apache.camel.karavan.service.InfinispanService;
-import org.apache.camel.karavan.service.KubernetesService;
import org.apache.camel.karavan.service.RunnerService;
import org.eclipse.microprofile.config.inject.ConfigProperty;
@@ -48,9 +48,6 @@ public class RunnerResource {
@Inject
RunnerService runnerServices;
- @Inject
- KubernetesService kubernetesService;
-
@Inject
InfinispanService infinispanService;
@@ -62,9 +59,9 @@ public class RunnerResource {
String runnerName = project.getProjectId() + "-" + RUNNER_SUFFIX;
String status = infinispanService.getRunnerStatus(runnerName,
RunnerStatus.NAME.context);
if (status == null) {
- Project p = infinispanService.getProject(project.getProjectId());
infinispanService.saveRunnerStatus(runnerName,
STATUS_NEED_INITIAL_LOAD, STATUS_NEED_INITIAL_LOAD);
- return Response.ok(kubernetesService.tryCreateRunner(p,
runnerName, jBangOptions)).build();
+ infinispanService.sendRunnerCommand(project.getProjectId(),
RunnerCommand.NAME.run);
+ return Response.ok(runnerName).build();
}
return Response.notModified().build();
}
@@ -89,9 +86,7 @@ public class RunnerResource {
@Consumes(MediaType.APPLICATION_JSON)
@Path("/{projectId}/{deletePVC}")
public Response deleteRunner(@PathParam("projectId") String projectId,
@PathParam("deletePVC") boolean deletePVC) {
- String runnerName = projectId + "-" + RUNNER_SUFFIX;
- kubernetesService.deleteRunner(runnerName, deletePVC);
- infinispanService.deleteRunnerStatuses(runnerName);
+ infinispanService.sendRunnerCommand(projectId,
RunnerCommand.NAME.delete);
return Response.accepted().build();
}
diff --git
a/karavan-app/src/main/java/org/apache/camel/karavan/listener/ClientRunnerListener.java
b/karavan-app/src/main/java/org/apache/camel/karavan/listener/ClientRunnerListener.java
new file mode 100644
index 00000000..b1aabaa6
--- /dev/null
+++
b/karavan-app/src/main/java/org/apache/camel/karavan/listener/ClientRunnerListener.java
@@ -0,0 +1,46 @@
+package org.apache.camel.karavan.listener;
+
+import org.apache.camel.karavan.model.GroupedKey;
+import org.apache.camel.karavan.model.RunnerCommand;
+import org.apache.camel.karavan.service.InfinispanService;
+import org.apache.camel.karavan.service.KubernetesService;
+import org.infinispan.client.hotrod.annotation.ClientCacheEntryCreated;
+import org.infinispan.client.hotrod.annotation.ClientCacheEntryModified;
+import org.infinispan.client.hotrod.annotation.ClientListener;
+import org.infinispan.client.hotrod.event.ClientCacheEntryCreatedEvent;
+import org.infinispan.client.hotrod.event.ClientCacheEntryModifiedEvent;
+
+import java.util.Objects;
+
+@ClientListener
+public class ClientRunnerListener extends RunnerListener {
+
+ public ClientRunnerListener(InfinispanService infinispanService,
KubernetesService kubernetesService) {
+ super(infinispanService, kubernetesService);
+ }
+
+ @ClientCacheEntryCreated
+ public void entryCreated(ClientCacheEntryCreatedEvent<GroupedKey> event) {
+ System.out.println("entryCreated");
+ String command = event.getKey().getKey();
+ String projectId = event.getKey().getGroup();
+ if (Objects.equals(command, RunnerCommand.NAME.run.name())) {
+ startRunner(projectId);
+ } else if (Objects.equals(command, RunnerCommand.NAME.delete.name())) {
+ stopRunner(projectId);
+ }
+ }
+
+ @ClientCacheEntryModified
+ public void entryModified(ClientCacheEntryModifiedEvent<GroupedKey> event)
{
+ System.out.println("entryModified");
+ String command = event.getKey().getKey();
+ String projectId = event.getKey().getGroup();
+ if (Objects.equals(command, RunnerCommand.NAME.run.name())) {
+ startRunner(projectId);
+ } else if (Objects.equals(command, RunnerCommand.NAME.delete.name())) {
+ stopRunner(projectId);
+ }
+ }
+
+}
\ No newline at end of file
diff --git
a/karavan-app/src/main/java/org/apache/camel/karavan/listener/LocalRunnerListener.java
b/karavan-app/src/main/java/org/apache/camel/karavan/listener/LocalRunnerListener.java
new file mode 100644
index 00000000..d7db9d7e
--- /dev/null
+++
b/karavan-app/src/main/java/org/apache/camel/karavan/listener/LocalRunnerListener.java
@@ -0,0 +1,48 @@
+package org.apache.camel.karavan.listener;
+
+import org.apache.camel.karavan.model.GroupedKey;
+import org.apache.camel.karavan.model.RunnerCommand;
+import org.apache.camel.karavan.service.InfinispanService;
+import org.apache.camel.karavan.service.KubernetesService;
+import org.infinispan.notifications.Listener;
+import org.infinispan.notifications.cachelistener.annotation.CacheEntryCreated;
+import
org.infinispan.notifications.cachelistener.annotation.CacheEntryModified;
+import org.infinispan.notifications.cachelistener.event.CacheEntryCreatedEvent;
+import
org.infinispan.notifications.cachelistener.event.CacheEntryModifiedEvent;
+
+import java.util.Objects;
+
+
+@Listener(primaryOnly = true)
+public class LocalRunnerListener extends RunnerListener {
+
+ public LocalRunnerListener(InfinispanService infinispanService,
KubernetesService kubernetesService) {
+ super(infinispanService, kubernetesService);
+ }
+
+ @CacheEntryCreated
+ public void entryCreated(CacheEntryCreatedEvent<GroupedKey, String> event)
{
+ if (!event.isPre()) {
+ String command = event.getKey().getKey();
+ String projectId = event.getKey().getGroup();
+ if (Objects.equals(command, RunnerCommand.NAME.run.name())) {
+ startRunner(projectId);
+ } else if (Objects.equals(command,
RunnerCommand.NAME.delete.name())) {
+ stopRunner(projectId);
+ }
+ }
+ }
+
+ @CacheEntryModified
+ public void entryModified(CacheEntryModifiedEvent<GroupedKey, String>
event) {
+ if (!event.isPre()) {
+ String command = event.getKey().getKey();
+ String projectId = event.getKey().getGroup();
+ if (Objects.equals(command, RunnerCommand.NAME.run.name())) {
+ startRunner(projectId);
+ } else if (Objects.equals(command,
RunnerCommand.NAME.delete.name())) {
+ stopRunner(projectId);
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git
a/karavan-app/src/main/java/org/apache/camel/karavan/listener/RunnerListener.java
b/karavan-app/src/main/java/org/apache/camel/karavan/listener/RunnerListener.java
new file mode 100644
index 00000000..15699939
--- /dev/null
+++
b/karavan-app/src/main/java/org/apache/camel/karavan/listener/RunnerListener.java
@@ -0,0 +1,35 @@
+package org.apache.camel.karavan.listener;
+
+import org.apache.camel.karavan.model.Project;
+import org.apache.camel.karavan.service.InfinispanService;
+import org.apache.camel.karavan.service.KubernetesService;
+
+import static org.apache.camel.karavan.service.RunnerService.RUNNER_SUFFIX;
+
+public class RunnerListener {
+
+ protected final InfinispanService infinispanService;
+
+ protected final KubernetesService kubernetesService;
+
+ public RunnerListener(InfinispanService infinispanService,
KubernetesService kubernetesService) {
+ this.infinispanService = infinispanService;
+ this.kubernetesService = kubernetesService;
+ }
+
+ protected void startRunner(String projectId) {
+ String runnerName = projectId + "-" + RUNNER_SUFFIX;
+ if (kubernetesService.inKubernetes()) {
+ Project p = infinispanService.getProject(projectId);
+ kubernetesService.tryCreateRunner(p, runnerName, "");
+ }
+ }
+
+ protected void stopRunner(String projectId) {
+ String runnerName = projectId + "-" + RUNNER_SUFFIX;
+ if (kubernetesService.inKubernetes()) {
+ kubernetesService.deleteRunner(runnerName, false);
+ infinispanService.deleteRunnerStatuses(runnerName);
+ }
+ }
+}
\ No newline at end of file
diff --git
a/karavan-app/src/main/java/org/apache/camel/karavan/model/GroupedKey.java
b/karavan-app/src/main/java/org/apache/camel/karavan/model/GroupedKey.java
index a50e5ced..3ab02eaf 100644
--- a/karavan-app/src/main/java/org/apache/camel/karavan/model/GroupedKey.java
+++ b/karavan-app/src/main/java/org/apache/camel/karavan/model/GroupedKey.java
@@ -22,6 +22,18 @@ public class GroupedKey {
}
+ public void setGroup(String group) {
+ this.group = group;
+ }
+
+ public String getKey() {
+ return key;
+ }
+
+ public void setKey(String key) {
+ this.key = key;
+ }
+
public String getGroup() {
return group;
}
diff --git
a/karavan-app/src/main/java/org/apache/camel/karavan/model/RunnerCommand.java
b/karavan-app/src/main/java/org/apache/camel/karavan/model/RunnerCommand.java
new file mode 100644
index 00000000..ccd2ce62
--- /dev/null
+++
b/karavan-app/src/main/java/org/apache/camel/karavan/model/RunnerCommand.java
@@ -0,0 +1,13 @@
+package org.apache.camel.karavan.model;
+
+public class RunnerCommand {
+
+ public enum NAME {
+ run,
+ delete,
+ reload
+ }
+
+ public static final String CACHE = "runner_commands";
+
+}
diff --git
a/karavan-app/src/main/java/org/apache/camel/karavan/service/InfinispanService.java
b/karavan-app/src/main/java/org/apache/camel/karavan/service/InfinispanService.java
index 75f47e17..08738cdb 100644
---
a/karavan-app/src/main/java/org/apache/camel/karavan/service/InfinispanService.java
+++
b/karavan-app/src/main/java/org/apache/camel/karavan/service/InfinispanService.java
@@ -17,16 +17,9 @@
package org.apache.camel.karavan.service;
import io.smallrye.mutiny.tuples.Tuple2;
-import org.apache.camel.karavan.model.CamelStatus;
-import org.apache.camel.karavan.model.DeploymentStatus;
-import org.apache.camel.karavan.model.Environment;
-import org.apache.camel.karavan.model.GroupedKey;
-import org.apache.camel.karavan.model.PipelineStatus;
-import org.apache.camel.karavan.model.PodStatus;
-import org.apache.camel.karavan.model.Project;
-import org.apache.camel.karavan.model.ProjectFile;
-import org.apache.camel.karavan.model.RunnerStatus;
-import org.apache.camel.karavan.model.ServiceStatus;
+import org.apache.camel.karavan.listener.ClientRunnerListener;
+import org.apache.camel.karavan.listener.LocalRunnerListener;
+import org.apache.camel.karavan.model.*;
import org.eclipse.microprofile.health.HealthCheck;
import org.eclipse.microprofile.health.HealthCheckResponse;
import org.eclipse.microprofile.health.Readiness;
@@ -38,7 +31,6 @@ import org.infinispan.commons.api.CacheContainerAdmin;
import org.infinispan.commons.configuration.StringConfiguration;
import org.infinispan.configuration.cache.CacheMode;
import org.infinispan.configuration.cache.ConfigurationBuilder;
-import org.infinispan.configuration.cache.SingleFileStoreConfigurationBuilder;
import org.infinispan.configuration.global.GlobalConfigurationBuilder;
import org.infinispan.manager.DefaultCacheManager;
import org.infinispan.query.dsl.QueryFactory;
@@ -48,11 +40,7 @@ import javax.enterprise.context.ApplicationScoped;
import javax.enterprise.inject.Default;
import javax.inject.Inject;
import java.time.Instant;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
@@ -74,10 +62,13 @@ public class InfinispanService implements HealthCheck {
private BasicCache<String, Environment> environments;
private BasicCache<String, String> commits;
private BasicCache<GroupedKey, String> runnerStatuses;
+ private BasicCache<GroupedKey, String> runnerCommands;
private final AtomicBoolean ready = new AtomicBoolean(false);
@Inject
- RemoteCacheManager cacheManager;
+ RemoteCacheManager remoteCacheManager;
+
+ DefaultCacheManager localCacheManager;
@Inject
CodeService codeService;
@@ -87,44 +78,54 @@ public class InfinispanService implements HealthCheck {
+ " <groups enabled=\"true\"/>"
+ "</distributed-cache>";
- private static final Logger LOGGER =
Logger.getLogger(KaravanService.class.getName());
+ private static final Logger LOGGER =
Logger.getLogger(InfinispanService.class.getName());
void start() {
- if (cacheManager == null) {
+ if (remoteCacheManager == null) {
LOGGER.info("InfinispanService is starting in local mode");
GlobalConfigurationBuilder global =
GlobalConfigurationBuilder.defaultClusteredBuilder();
- DefaultCacheManager cacheManager = new
DefaultCacheManager(global.build());
+ localCacheManager = new DefaultCacheManager(global.build());
ConfigurationBuilder builder = new ConfigurationBuilder();
builder.clustering().cacheMode(CacheMode.LOCAL);
- environments =
cacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache(Environment.CACHE,
builder.build());
- projects =
cacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache(Project.CACHE,
builder.build());
- files =
cacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache(ProjectFile.CACHE,
builder.build());
- pipelineStatuses =
cacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache(PipelineStatus.CACHE,
builder.build());
- deploymentStatuses =
cacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache(DeploymentStatus.CACHE,
builder.build());
- podStatuses =
cacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache(PodStatus.CACHE,
builder.build());
- serviceStatuses =
cacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache(ServiceStatus.CACHE,
builder.build());
- camelStatuses =
cacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache(CamelStatus.CACHE,
builder.build());
- commits =
cacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache("commits",
builder.build());
- runnerStatuses =
cacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache("runner_statuses",
builder.build());
+ environments =
localCacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache(Environment.CACHE,
builder.build());
+ projects =
localCacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache(Project.CACHE,
builder.build());
+ files =
localCacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache(ProjectFile.CACHE,
builder.build());
+ pipelineStatuses =
localCacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache(PipelineStatus.CACHE,
builder.build());
+ deploymentStatuses =
localCacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache(DeploymentStatus.CACHE,
builder.build());
+ podStatuses =
localCacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache(PodStatus.CACHE,
builder.build());
+ serviceStatuses =
localCacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache(ServiceStatus.CACHE,
builder.build());
+ camelStatuses =
localCacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache(CamelStatus.CACHE,
builder.build());
+ commits =
localCacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache("commits",
builder.build());
+ runnerStatuses =
localCacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache("runner_statuses",
builder.build());
+ runnerCommands =
localCacheManager.administration().withFlags(CacheContainerAdmin.AdminFlag.VOLATILE).getOrCreateCache(RunnerCommand.CACHE,
builder.build());
cleanData();
} else {
LOGGER.info("InfinispanService is starting in remote mode");
- environments =
cacheManager.administration().getOrCreateCache(Environment.CACHE, new
StringConfiguration(String.format(CACHE_CONFIG, Environment.CACHE)));
- projects =
cacheManager.administration().getOrCreateCache(Project.CACHE, new
StringConfiguration(String.format(CACHE_CONFIG, Project.CACHE)));
- files =
cacheManager.administration().getOrCreateCache(ProjectFile.CACHE, new
StringConfiguration(String.format(CACHE_CONFIG, ProjectFile.CACHE)));
- pipelineStatuses =
cacheManager.administration().getOrCreateCache(PipelineStatus.CACHE, new
StringConfiguration(String.format(CACHE_CONFIG, PipelineStatus.CACHE)));
- deploymentStatuses =
cacheManager.administration().getOrCreateCache(DeploymentStatus.CACHE, new
StringConfiguration(String.format(CACHE_CONFIG, DeploymentStatus.CACHE)));
- podStatuses =
cacheManager.administration().getOrCreateCache(PodStatus.CACHE, new
StringConfiguration(String.format(CACHE_CONFIG, PodStatus.CACHE)));
- serviceStatuses =
cacheManager.administration().getOrCreateCache(ServiceStatus.CACHE, new
StringConfiguration(String.format(CACHE_CONFIG, ServiceStatus.CACHE)));
- camelStatuses =
cacheManager.administration().getOrCreateCache(CamelStatus.CACHE, new
StringConfiguration(String.format(CACHE_CONFIG, CamelStatus.CACHE)));
- commits =
cacheManager.administration().getOrCreateCache("commits", new
StringConfiguration(String.format(CACHE_CONFIG, "commits")));
- runnerStatuses =
cacheManager.administration().getOrCreateCache("runner_statuses", new
StringConfiguration(String.format(CACHE_CONFIG, "runner_statuses")));
+ environments =
remoteCacheManager.administration().getOrCreateCache(Environment.CACHE, new
StringConfiguration(String.format(CACHE_CONFIG, Environment.CACHE)));
+ projects =
remoteCacheManager.administration().getOrCreateCache(Project.CACHE, new
StringConfiguration(String.format(CACHE_CONFIG, Project.CACHE)));
+ files =
remoteCacheManager.administration().getOrCreateCache(ProjectFile.CACHE, new
StringConfiguration(String.format(CACHE_CONFIG, ProjectFile.CACHE)));
+ pipelineStatuses =
remoteCacheManager.administration().getOrCreateCache(PipelineStatus.CACHE, new
StringConfiguration(String.format(CACHE_CONFIG, PipelineStatus.CACHE)));
+ deploymentStatuses =
remoteCacheManager.administration().getOrCreateCache(DeploymentStatus.CACHE,
new StringConfiguration(String.format(CACHE_CONFIG, DeploymentStatus.CACHE)));
+ podStatuses =
remoteCacheManager.administration().getOrCreateCache(PodStatus.CACHE, new
StringConfiguration(String.format(CACHE_CONFIG, PodStatus.CACHE)));
+ serviceStatuses =
remoteCacheManager.administration().getOrCreateCache(ServiceStatus.CACHE, new
StringConfiguration(String.format(CACHE_CONFIG, ServiceStatus.CACHE)));
+ camelStatuses =
remoteCacheManager.administration().getOrCreateCache(CamelStatus.CACHE, new
StringConfiguration(String.format(CACHE_CONFIG, CamelStatus.CACHE)));
+ commits =
remoteCacheManager.administration().getOrCreateCache("commits", new
StringConfiguration(String.format(CACHE_CONFIG, "commits")));
+ runnerStatuses =
remoteCacheManager.administration().getOrCreateCache("runner_statuses", new
StringConfiguration(String.format(CACHE_CONFIG, "runner_statuses")));
+ runnerCommands =
remoteCacheManager.administration().getOrCreateCache(RunnerCommand.CACHE, new
StringConfiguration(String.format(CACHE_CONFIG, RunnerCommand.CACHE)));
}
+ addListeners();
ready.set(true);
}
- public RemoteCacheManager getRemoteCacheManager() {
- return cacheManager;
+ @Inject
+ KubernetesService kubernetesService;
+
+ private void addListeners() {
+ if (remoteCacheManager != null) {
+
remoteCacheManager.getCache(RunnerCommand.CACHE).addClientListener(new
ClientRunnerListener(this, kubernetesService));
+ } else {
+ localCacheManager.getCache(RunnerCommand.CACHE).addListener(new
LocalRunnerListener(this, kubernetesService));
+ }
}
private void cleanData() {
@@ -152,7 +153,7 @@ public class InfinispanService implements HealthCheck {
}
public List<ProjectFile> getProjectFiles(String projectId) {
- if (cacheManager == null) {
+ if (remoteCacheManager == null) {
return files.values().stream()
.filter(f -> f.getProjectId().equals(projectId))
.collect(Collectors.toList());
@@ -165,7 +166,7 @@ public class InfinispanService implements HealthCheck {
}
public ProjectFile getProjectFile(String projectId, String filename) {
- if (cacheManager == null) {
+ if (remoteCacheManager == null) {
return files.values().stream()
.filter(f -> f.getProjectId().equals(projectId) &&
f.getName().equals(filename))
.findFirst().orElse(new ProjectFile());
@@ -232,7 +233,7 @@ public class InfinispanService implements HealthCheck {
}
public List<DeploymentStatus> getDeploymentStatuses(String env) {
- if (cacheManager == null) {
+ if (remoteCacheManager == null) {
return deploymentStatuses.values().stream()
.filter(s -> s.getEnv().equals(env))
.collect(Collectors.toList());
@@ -257,7 +258,7 @@ public class InfinispanService implements HealthCheck {
}
public List<PodStatus> getPodStatuses(String projectId, String env) {
- if (cacheManager == null) {
+ if (remoteCacheManager == null) {
return podStatuses.values().stream()
.filter(s -> s.getEnv().equals(env) &&
s.getProject().equals(projectId))
.collect(Collectors.toList());
@@ -271,7 +272,7 @@ public class InfinispanService implements HealthCheck {
}
public List<PodStatus> getPodStatuses(String env) {
- if (cacheManager == null) {
+ if (remoteCacheManager == null) {
return podStatuses.values().stream()
.filter(s -> s.getEnv().equals(env))
.collect(Collectors.toList());
@@ -296,7 +297,7 @@ public class InfinispanService implements HealthCheck {
}
public List<CamelStatus> getCamelStatuses(String env) {
- if (cacheManager == null) {
+ if (remoteCacheManager == null) {
return camelStatuses.values().stream()
.filter(s -> s.getEnv().equals(env))
.collect(Collectors.toList());
@@ -368,13 +369,17 @@ public class InfinispanService implements HealthCheck {
return commits.get(commitId) != null;
}
+ public void sendRunnerCommand(String projectId, RunnerCommand.NAME
command) {
+ runnerCommands.put(GroupedKey.create(projectId, command.name()),
UUID.randomUUID().toString());
+ }
+
@Override
public HealthCheckResponse call() {
- if (cacheManager == null && ready.get()){
+ if (remoteCacheManager == null && ready.get()){
return HealthCheckResponse.up("Infinispan Service is running in
local mode.");
}
else {
- if (cacheManager != null && cacheManager.isStarted() &&
ready.get()) {
+ if (remoteCacheManager != null && remoteCacheManager.isStarted()
&& ready.get()) {
return HealthCheckResponse.up("Infinispan Service is running
in cluster mode.");
}
else {
diff --git
a/karavan-app/src/main/java/org/apache/camel/karavan/service/KubernetesService.java
b/karavan-app/src/main/java/org/apache/camel/karavan/service/KubernetesService.java
index cc538db7..69203f3b 100644
---
a/karavan-app/src/main/java/org/apache/camel/karavan/service/KubernetesService.java
+++
b/karavan-app/src/main/java/org/apache/camel/karavan/service/KubernetesService.java
@@ -51,7 +51,7 @@ import static
org.apache.camel.karavan.service.ServiceUtil.APPLICATION_PROPERTIE
@Default
@Readiness
@ApplicationScoped
-public class KubernetesService implements HealthCheck{
+public class KubernetesService implements HealthCheck {
private static final Logger LOGGER =
Logger.getLogger(KubernetesService.class.getName());
public static final String START_INFORMERS = "start-informers";
@@ -103,22 +103,22 @@ public class KubernetesService implements HealthCheck{
SharedIndexInformer<Deployment> deploymentInformer =
kubernetesClient().apps().deployments().inNamespace(getNamespace())
.withLabels(getRuntimeLabels()).inform();
- deploymentInformer.addEventHandlerWithResyncPeriod(new
DeploymentEventHandler(infinispanService, this),30 * 1000L);
+ deploymentInformer.addEventHandlerWithResyncPeriod(new
DeploymentEventHandler(infinispanService, this), 30 * 1000L);
informers.add(deploymentInformer);
SharedIndexInformer<Service> serviceInformer =
kubernetesClient().services().inNamespace(getNamespace())
.withLabels(getRuntimeLabels()).inform();
- serviceInformer.addEventHandlerWithResyncPeriod(new
ServiceEventHandler(infinispanService, this),30 * 1000L);
+ serviceInformer.addEventHandlerWithResyncPeriod(new
ServiceEventHandler(infinispanService, this), 30 * 1000L);
informers.add(serviceInformer);
SharedIndexInformer<PipelineRun> pipelineRunInformer =
tektonClient().v1beta1().pipelineRuns().inNamespace(getNamespace())
.withLabels(getRuntimeLabels()).inform();
- pipelineRunInformer.addEventHandlerWithResyncPeriod(new
PipelineRunEventHandler(infinispanService, this),30 * 1000L);
+ pipelineRunInformer.addEventHandlerWithResyncPeriod(new
PipelineRunEventHandler(infinispanService, this), 30 * 1000L);
informers.add(pipelineRunInformer);
SharedIndexInformer<Pod> podRunInformer =
kubernetesClient().pods().inNamespace(getNamespace())
.withLabels(getRuntimeLabels()).inform();
- podRunInformer.addEventHandlerWithResyncPeriod(new
PodEventHandler(infinispanService, this),30 * 1000L);
+ podRunInformer.addEventHandlerWithResyncPeriod(new
PodEventHandler(infinispanService, this), 30 * 1000L);
informers.add(podRunInformer);
LOGGER.info("Started Kubernetes Informers");
@@ -127,14 +127,17 @@ public class KubernetesService implements HealthCheck{
}
}
-
+
@Override
public HealthCheckResponse call() {
- if(informers.size() == INFORMERS) {
- return HealthCheckResponse.up("All Kubernetes informers are
running.");
- }
- else {
- return HealthCheckResponse.down("kubernetes Informers are not
running.");
+ if (inKubernetes()) {
+ if (informers.size() == INFORMERS) {
+ return HealthCheckResponse.up("All Kubernetes informers are
running.");
+ } else {
+ return HealthCheckResponse.down("kubernetes Informers are not
running.");
+ }
+ } else {
+ return HealthCheckResponse.up("Running Kubernetesless.");
}
}
@@ -155,7 +158,7 @@ public class KubernetesService implements HealthCheck{
Map<String, String> labels = getRuntimeLabels(
Map.of("karavan-project-id", project.getProjectId(),
- "tekton.dev/pipeline", pipeline)
+ "tekton.dev/pipeline", pipeline)
);
ObjectMeta meta = new ObjectMetaBuilder()
@@ -250,7 +253,7 @@ public class KubernetesService implements HealthCheck{
}
}
- private List<Condition> getCancelConditions(String reason){
+ private List<Condition> getCancelConditions(String reason) {
List<Condition> cancelConditions = new ArrayList<>();
Condition taskRunCancelCondition = new Condition();
taskRunCancelCondition.setType("Succeeded");
@@ -260,7 +263,7 @@ public class KubernetesService implements HealthCheck{
cancelConditions.add(taskRunCancelCondition);
return cancelConditions;
}
-
+
public void rolloutDeployment(String name, String namespace) {
try {
kubernetesClient().apps().deployments().inNamespace(namespace).withName(name).rolling().restart();
@@ -328,12 +331,12 @@ public class KubernetesService implements HealthCheck{
public List<String> getConfigMaps(String namespace) {
List<String> result = new ArrayList<>();
try {
-
kubernetesClient().configMaps().inNamespace(namespace).list().getItems().forEach(configMap
-> {
- String name = configMap.getMetadata().getName();
- if (configMap.getData() != null) {
- configMap.getData().keySet().forEach(data -> result.add(name +
"/" + data));
- }
- });
+
kubernetesClient().configMaps().inNamespace(namespace).list().getItems().forEach(configMap
-> {
+ String name = configMap.getMetadata().getName();
+ if (configMap.getData() != null) {
+ configMap.getData().keySet().forEach(data ->
result.add(name + "/" + data));
+ }
+ });
} catch (Exception e) {
LOGGER.error(e);
}
@@ -358,11 +361,11 @@ public class KubernetesService implements HealthCheck{
public List<String> getServices(String namespace) {
List<String> result = new ArrayList<>();
try {
-
kubernetesClient().services().inNamespace(namespace).list().getItems().forEach(service
-> {
- String name = service.getMetadata().getName();
- String host = name + "." + namespace + ".svc.cluster.local";
- service.getSpec().getPorts().forEach(port -> result.add(name + "|"
+ host + ":" + port.getPort()));
- });
+
kubernetesClient().services().inNamespace(namespace).list().getItems().forEach(service
-> {
+ String name = service.getMetadata().getName();
+ String host = name + "." + namespace + ".svc.cluster.local";
+ service.getSpec().getPorts().forEach(port -> result.add(name +
"|" + host + ":" + port.getPort()));
+ });
} catch (Exception e) {
LOGGER.error(e);
}
@@ -391,7 +394,7 @@ public class KubernetesService implements HealthCheck{
Pod old =
kubernetesClient().pods().inNamespace(getNamespace()).withName(runnerName).get();
if (old == null) {
ProjectFile properties =
infinispanService.getProjectFile(project.getProjectId(),
APPLICATION_PROPERTIES_FILENAME);
- Map<String,String> containerResources = ServiceUtil
+ Map<String, String> containerResources = ServiceUtil
.getRunnerContainerResourcesMap(properties, isOpenshift(),
project.getRuntime().equals("quarkus"));
Pod pod = getRunnerPod(project.getProjectId(), runnerName,
jBangOptions, containerResources);
Pod result = kubernetesClient().resource(pod).createOrReplace();
@@ -414,7 +417,7 @@ public class KubernetesService implements HealthCheck{
}
}
- public ResourceRequirements getResourceRequirements(Map<String,String>
containerResources) {
+ public ResourceRequirements getResourceRequirements(Map<String, String>
containerResources) {
return new ResourceRequirementsBuilder()
.addToRequests("cpu", new
Quantity(containerResources.get("requests.cpu")))
.addToRequests("memory", new
Quantity(containerResources.get("requests.memory")))
@@ -423,8 +426,8 @@ public class KubernetesService implements HealthCheck{
.build();
}
- private Pod getRunnerPod(String projectId, String name, String
jbangOptions, Map<String,String> containerResources) {
- Map<String,String> labels = new HashMap<>();
+ private Pod getRunnerPod(String projectId, String name, String
jbangOptions, Map<String, String> containerResources) {
+ Map<String, String> labels = new HashMap<>();
labels.putAll(getRuntimeLabels());
labels.putAll(getKaravanRunnerLabels(name));
labels.put("karavan/projectId", projectId);
@@ -534,7 +537,7 @@ public class KubernetesService implements HealthCheck{
}
public static Map<String, String> getKaravanRunnerLabels(String name) {
- return Map.of("karavan/type" , "runner",
+ return Map.of("karavan/type", "runner",
"app.kubernetes.io/name", name);
}
@@ -545,6 +548,7 @@ public class KubernetesService implements HealthCheck{
public String getCluster() {
return kubernetesClient().getMasterUrl().getHost();
}
+
public String getNamespace() {
return currentNamespace;
}
diff --git
a/karavan-app/src/main/java/org/apache/camel/karavan/service/RunnerService.java
b/karavan-app/src/main/java/org/apache/camel/karavan/service/RunnerService.java
index 0c5861f9..a0eefe52 100644
---
a/karavan-app/src/main/java/org/apache/camel/karavan/service/RunnerService.java
+++
b/karavan-app/src/main/java/org/apache/camel/karavan/service/RunnerService.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.karavan.service;
+import io.fabric8.kubernetes.api.model.Pod;
import io.quarkus.scheduler.Scheduled;
import io.quarkus.vertx.ConsumeEvent;
import io.vertx.core.json.JsonObject;
@@ -25,6 +26,8 @@ import io.vertx.mutiny.core.eventbus.EventBus;
import io.vertx.mutiny.ext.web.client.HttpResponse;
import io.vertx.mutiny.ext.web.client.WebClient;
import org.apache.camel.karavan.model.PodStatus;
+import org.apache.camel.karavan.model.Project;
+import org.apache.camel.karavan.model.ProjectFile;
import org.apache.camel.karavan.model.RunnerStatus;
import org.eclipse.microprofile.config.inject.ConfigProperty;
import org.eclipse.microprofile.faulttolerance.CircuitBreaker;
@@ -33,9 +36,12 @@ import org.jboss.logging.Logger;
import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;
import java.util.Arrays;
+import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ExecutionException;
+import static
org.apache.camel.karavan.service.ServiceUtil.APPLICATION_PROPERTIES_FILENAME;
+
@ApplicationScoped
public class RunnerService {
@@ -83,7 +89,7 @@ public class RunnerService {
@CircuitBreaker(requestVolumeThreshold = 10, failureRatio = 0.5, delay =
1000)
public boolean putRequest(String runnerName, String fileName, String body,
int timeout) {
try {
- String url = "http://"; + runnerName + "." +
kubernetesService.getNamespace() + ".svc.cluster.local/q/upload/" + fileName;
+ String url = getRunnerAddress(runnerName) + "/q/upload/" +
fileName;
HttpResponse<Buffer> result = getWebClient().putAbs(url)
.timeout(timeout).sendBuffer(Buffer.buffer(body)).subscribeAsCompletionStage().toCompletableFuture().get();
return result.statusCode() == 200;
@@ -94,7 +100,7 @@ public class RunnerService {
}
public String reloadRequest(String runnerName) {
- String url = "http://"; + runnerName + "." +
kubernetesService.getNamespace() +
".svc.cluster.local/q/dev/reload?reload=true";
+ String url = getRunnerAddress(runnerName) +
"/q/dev/reload?reload=true";
try {
return result(url, 1000);
} catch (InterruptedException | ExecutionException e) {
@@ -103,8 +109,17 @@ public class RunnerService {
return null;
}
+ public String getRunnerAddress(String runnerName) {
+ if (kubernetesService.inKubernetes()) {
+ return "http://"; + runnerName + "." +
kubernetesService.getNamespace() + ".svc.cluster.local";
+ } else {
+ return "http://"; + runnerName + ":8080";
+ }
+ }
+
@Scheduled(every = "{karavan.runner-status-interval}", concurrentExecution
= Scheduled.ConcurrentExecution.SKIP)
void collectRunnerStatus() {
+ System.out.println("collectRunnerStatus");
if (infinispanService.call().getStatus().name().equals("UP")) {
infinispanService.getPodStatuses(environment).stream().filter(PodStatus::getRunner).forEach(podStatus
-> {
eventBus.publish(CMD_COLLECT_RUNNER_STATUS,
podStatus.getName());
@@ -151,8 +166,7 @@ public class RunnerService {
}
public String getRunnerStatus(String podName, RunnerStatus.NAME
statusName) {
- String url = "http://"; + podName + "." +
kubernetesService.getNamespace() + ".svc.cluster.local/q/dev/" +
statusName.name();
-// String url = "http://0.0.0.0:8888/q/dev/"; + statusName.name();
+ String url = getRunnerAddress(podName) + "/q/dev/" + statusName.name();
try {
return result(url, 1000);
} catch (InterruptedException | ExecutionException e) {
diff --git a/karavan-app/src/main/resources/application.properties
b/karavan-app/src/main/resources/application.properties
index 741210d8..414ffac6 100644
--- a/karavan-app/src/main/resources/application.properties
+++ b/karavan-app/src/main/resources/application.properties
@@ -19,8 +19,6 @@ quarkus.infinispan-client.username=admin
quarkus.infinispan-client.password=karavan
quarkus.infinispan-client.devservices.enabled=false
-quarkus.infinispan-client.devservices.port=12345
-quarkus.infinispan-client.devservices.service-name=karavan
quarkus.infinispan-client.health.enabled=false
# Infinispan client intelligence
diff --git a/karavan-bashi/pom.xml b/karavan-bashi/pom.xml
index a5ae14d4..f13a3ce5 100644
--- a/karavan-bashi/pom.xml
+++ b/karavan-bashi/pom.xml
@@ -41,6 +41,10 @@
<groupId>io.quarkus</groupId>
<artifactId>quarkus-vertx</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.quarkus</groupId>
+ <artifactId>quarkus-infinispan-client</artifactId>
+ </dependency>
<dependency>
<groupId>com.github.docker-java</groupId>
<artifactId>docker-java-core</artifactId>
diff --git
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/KaravanBashi.java
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/ConductorService.java
similarity index 61%
copy from
karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/KaravanBashi.java
copy to
karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/ConductorService.java
index 3058c2a0..12ebdb8b 100644
---
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/KaravanBashi.java
+++
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/ConductorService.java
@@ -1,22 +1,21 @@
package org.apache.camel.karavan.bashi;
import com.github.dockerjava.api.model.HealthCheck;
-import io.quarkus.runtime.ShutdownEvent;
-import io.quarkus.runtime.StartupEvent;
import io.quarkus.vertx.ConsumeEvent;
+import io.vertx.core.json.JsonObject;
import org.apache.camel.karavan.bashi.docker.DockerService;
import org.eclipse.microprofile.config.inject.ConfigProperty;
import org.jboss.logging.Logger;
import javax.enterprise.context.ApplicationScoped;
-import javax.enterprise.event.Observes;
import javax.inject.Inject;
import java.util.List;
+import java.util.Map;
-import static org.apache.camel.karavan.bashi.KaravanConstants.*;
+import static org.apache.camel.karavan.bashi.Constants.*;
@ApplicationScoped
-public class KaravanBashi {
+public class ConductorService {
@ConfigProperty(name = "karavan.image")
String karavanImage;
@@ -30,6 +29,8 @@ public class KaravanBashi {
String gitPassword;
@ConfigProperty(name = "karavan.git-branch")
String gitBranch;
+ @ConfigProperty(name = "karavan.runner-image")
+ String runnerImage;
@ConfigProperty(name = "infinispan.image")
String infinispanImage;
@@ -43,17 +44,14 @@ public class KaravanBashi {
@Inject
DockerService dockerService;
- private static final Logger LOGGER =
Logger.getLogger(KaravanBashi.class.getName());
+ private static final Logger LOGGER =
Logger.getLogger(ConductorService.class.getName());
- void onStart(@Observes StartupEvent ev) throws InterruptedException {
- LOGGER.info("Karavan Bashi is starting...");
- dockerService.checkContainersStatus();
- dockerService.createNetwork();
- dockerService.startListeners();
- startInfinispan();
- }
+ public static final String ADDRESS_INFINISPAN_START =
"ADDRESS_INFINISPAN_START";
+ public static final String ADDRESS_INFINISPAN_HEALTH =
"ADDRESS_INFINISPAN_HEALTH";
+ public static final String ADDRESS_RUNNER = "ADDRESS_RUNNER";
- void startInfinispan() throws InterruptedException {
+ @ConsumeEvent(value = ADDRESS_INFINISPAN_START, blocking = true, ordered =
true)
+ void startInfinispan(String data) throws InterruptedException {
LOGGER.info("Infinispan is starting...");
HealthCheck healthCheck = new HealthCheck().withTest(List.of("CMD",
"curl", "-f",
"http://localhost:11222/rest/v2/cache-managers/default/health/status";))
@@ -61,13 +59,13 @@ public class KaravanBashi {
dockerService.createContainer(INFINISPAN_CONTAINER_NAME,
infinispanImage,
List.of("USER=" + infinispanUsername, "PASS=" +
infinispanPassword),
- infinispanPort, true, healthCheck
+ infinispanPort, true, healthCheck, Map.of()
);
dockerService.startContainer(INFINISPAN_CONTAINER_NAME);
LOGGER.info("Infinispan is started");
}
- @ConsumeEvent(value = ADDRESS_INFINISPAN_HEALTH, blocking = true, ordered
= false)
+ @ConsumeEvent(value = ADDRESS_INFINISPAN_HEALTH, blocking = true, ordered
= true)
void startKaravan(String infinispanHealth) throws InterruptedException {
if (infinispanHealth.equals("healthy")) {
LOGGER.info("Karavan is starting...");
@@ -79,14 +77,28 @@ public class KaravanBashi {
"KARAVAN_GIT_PASSWORD=" + gitPassword,
"KARAVAN_GIT_BRANCH=" + gitBranch
),
- karavanPort, true, new HealthCheck()
+ karavanPort, true, new HealthCheck(), Map.of()
);
dockerService.startContainer(KARAVAN_CONTAINER_NAME);
LOGGER.info("Karavan is started");
}
}
- void onStop(@Observes ShutdownEvent ev) {
- LOGGER.info("Karavan Bashi is stopping...");
+ @ConsumeEvent(value = ADDRESS_RUNNER, blocking = true, ordered = true)
+ void manageRunner(JsonObject params) throws InterruptedException {
+ String projectId = params.getString("projectId");
+ String command = params.getString("command");
+ String runnerName = projectId + "-" + RUNNER_SUFFIX;
+ if (command.equals("run")) {
+ LOGGER.infof("Runner starting for %s", projectId);
+ dockerService.createContainer(runnerName, runnerImage,
+ List.of(), "", false, new HealthCheck(), Map.of("type",
"runner")
+ );
+ dockerService.startContainer(runnerName);
+ LOGGER.infof("Runner started for %s", projectId);
+ } else {
+ dockerService.stopContainer(runnerName);
+ dockerService.deleteContainer(runnerName);
+ }
}
}
diff --git
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/KaravanConstants.java
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/Constants.java
similarity index 66%
rename from
karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/KaravanConstants.java
rename to
karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/Constants.java
index 8e53e011..e9115f67 100644
---
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/KaravanConstants.java
+++ b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/Constants.java
@@ -1,11 +1,10 @@
package org.apache.camel.karavan.bashi;
-public class KaravanConstants {
+public class Constants {
public static final String NETWORK_NAME = "karavan";
public static final String INFINISPAN_CONTAINER_NAME = "infinispan";
public static final String KARAVAN_CONTAINER_NAME = "karavan";
-
- public static final String ADDRESS_INFINISPAN_HEALTH =
"ADDRESS_INFINISPAN_HEALTH";
+ public static final String RUNNER_SUFFIX = "runner";
}
diff --git
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/HealthChecker.java
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/HealthChecker.java
deleted file mode 100644
index 681fbf19..00000000
---
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/HealthChecker.java
+++ /dev/null
@@ -1,31 +0,0 @@
-package org.apache.camel.karavan.bashi;
-
-import com.github.dockerjava.api.model.Container;
-import io.quarkus.scheduler.Scheduled;
-import org.jboss.logging.Logger;
-
-import javax.enterprise.context.ApplicationScoped;
-import java.util.concurrent.ConcurrentHashMap;
-
-@ApplicationScoped
-public class HealthChecker {
-
- private static final Logger LOGGER =
Logger.getLogger(HealthChecker.class.getName());
-
- private static final ConcurrentHashMap<String, Container> containers = new
ConcurrentHashMap<>();
-
-// @Scheduled(every = "{karavan.health-checker-interval}",
concurrentExecution = Scheduled.ConcurrentExecution.SKIP)
-// void collectHealthStatuses() {
-// containers.forEach((s, s2) -> {
-// LOGGER.infof("HealthCheck for %s", s);
-// });
-// }
-
-// public void addContainer(Container container){
-// containers.put(container.getId(), container);
-// }
-//
-// public void removeContainer(String id){
-// containers.remove(id);
-// }
-}
diff --git
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/KaravanBashi.java
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/KaravanBashi.java
index 3058c2a0..7b3ed5d7 100644
---
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/KaravanBashi.java
+++
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/KaravanBashi.java
@@ -1,44 +1,20 @@
package org.apache.camel.karavan.bashi;
-import com.github.dockerjava.api.model.HealthCheck;
import io.quarkus.runtime.ShutdownEvent;
import io.quarkus.runtime.StartupEvent;
-import io.quarkus.vertx.ConsumeEvent;
+import io.vertx.core.eventbus.EventBus;
import org.apache.camel.karavan.bashi.docker.DockerService;
-import org.eclipse.microprofile.config.inject.ConfigProperty;
import org.jboss.logging.Logger;
import javax.enterprise.context.ApplicationScoped;
import javax.enterprise.event.Observes;
import javax.inject.Inject;
-import java.util.List;
-
-import static org.apache.camel.karavan.bashi.KaravanConstants.*;
@ApplicationScoped
public class KaravanBashi {
- @ConfigProperty(name = "karavan.image")
- String karavanImage;
- @ConfigProperty(name = "karavan.port")
- String karavanPort;
- @ConfigProperty(name = "karavan.git-repository")
- String gitRepository;
- @ConfigProperty(name = "karavan.git-username")
- String gitUsername;
- @ConfigProperty(name = "karavan.git-password")
- String gitPassword;
- @ConfigProperty(name = "karavan.git-branch")
- String gitBranch;
-
- @ConfigProperty(name = "infinispan.image")
- String infinispanImage;
- @ConfigProperty(name = "infinispan.port")
- String infinispanPort;
- @ConfigProperty(name = "infinispan.username")
- String infinispanUsername;
- @ConfigProperty(name = "infinispan.password")
- String infinispanPassword;
+ @Inject
+ EventBus eventBus;
@Inject
DockerService dockerService;
@@ -50,40 +26,7 @@ public class KaravanBashi {
dockerService.checkContainersStatus();
dockerService.createNetwork();
dockerService.startListeners();
- startInfinispan();
- }
-
- void startInfinispan() throws InterruptedException {
- LOGGER.info("Infinispan is starting...");
-
- HealthCheck healthCheck = new HealthCheck().withTest(List.of("CMD",
"curl", "-f",
"http://localhost:11222/rest/v2/cache-managers/default/health/status";))
-
.withInterval(10000000000L).withTimeout(10000000000L).withStartPeriod(10000000000L).withRetries(30);
-
- dockerService.createContainer(INFINISPAN_CONTAINER_NAME,
infinispanImage,
- List.of("USER=" + infinispanUsername, "PASS=" +
infinispanPassword),
- infinispanPort, true, healthCheck
- );
- dockerService.startContainer(INFINISPAN_CONTAINER_NAME);
- LOGGER.info("Infinispan is started");
- }
-
- @ConsumeEvent(value = ADDRESS_INFINISPAN_HEALTH, blocking = true, ordered
= false)
- void startKaravan(String infinispanHealth) throws InterruptedException {
- if (infinispanHealth.equals("healthy")) {
- LOGGER.info("Karavan is starting...");
- dockerService.createContainer(KARAVAN_CONTAINER_NAME, karavanImage,
- List.of(
- "QUARKUS_INFINISPAN_CLIENT_HOSTS=infinispan:11222",
- "KARAVAN_GIT_REPOSITORY=" + gitRepository,
- "KARAVAN_GIT_USERNAME=" + gitUsername,
- "KARAVAN_GIT_PASSWORD=" + gitPassword,
- "KARAVAN_GIT_BRANCH=" + gitBranch
- ),
- karavanPort, true, new HealthCheck()
- );
- dockerService.startContainer(KARAVAN_CONTAINER_NAME);
- LOGGER.info("Karavan is started");
- }
+ eventBus.publish(ConductorService.ADDRESS_INFINISPAN_START, "");
}
void onStop(@Observes ShutdownEvent ev) {
diff --git
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/KaravanContainers.java
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/KaravanContainers.java
deleted file mode 100644
index 7fe14977..00000000
---
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/KaravanContainers.java
+++ /dev/null
@@ -1,33 +0,0 @@
-package org.apache.camel.karavan.bashi;
-
-import com.github.dockerjava.api.model.Container;
-import io.vertx.core.eventbus.EventBus;
-import org.jboss.logging.Logger;
-
-import javax.enterprise.context.ApplicationScoped;
-import javax.inject.Inject;
-import java.util.concurrent.ConcurrentHashMap;
-
-import static
org.apache.camel.karavan.bashi.KaravanConstants.ADDRESS_INFINISPAN_HEALTH;
-
-@ApplicationScoped
-public class KaravanContainers {
-
- private static final Logger LOGGER =
Logger.getLogger(KaravanContainers.class.getName());
-
- private static final ConcurrentHashMap<String, String> containers = new
ConcurrentHashMap<>();
-
- @Inject
- EventBus eventBus;
-
- public void addContainer(Container container, String health){
- containers.put(container.getId(), health);
- if (container.getNames()[0].equals("/infinispan")) {
- eventBus.publish(ADDRESS_INFINISPAN_HEALTH, health);
- }
- }
-
- public void removeContainer(String id){
- containers.remove(id);
- }
-}
diff --git
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/RunnerStatusService.java
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/RunnerStatusService.java
new file mode 100644
index 00000000..038d40c5
--- /dev/null
+++
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/RunnerStatusService.java
@@ -0,0 +1,83 @@
+package org.apache.camel.karavan.bashi;
+
+import com.github.dockerjava.api.model.HealthCheck;
+import com.github.dockerjava.api.model.Statistics;
+import io.quarkus.scheduler.Scheduled;
+import io.quarkus.vertx.ConsumeEvent;
+import io.vertx.core.json.JsonObject;
+import org.apache.camel.karavan.bashi.docker.DockerService;
+import org.apache.camel.karavan.bashi.infinispan.InfinispanService;
+import org.apache.camel.karavan.bashi.infinispan.PodStatus;
+import org.eclipse.microprofile.config.inject.ConfigProperty;
+import org.jboss.logging.Logger;
+
+import javax.enterprise.context.ApplicationScoped;
+import javax.inject.Inject;
+import java.util.*;
+
+import static org.apache.camel.karavan.bashi.Constants.*;
+
+@ApplicationScoped
+public class RunnerStatusService {
+
+ @ConfigProperty(name = "karavan.environment")
+ String environment;
+
+ @Inject
+ DockerService dockerService;
+
+ @Inject
+ InfinispanService infinispanService;
+
+ private static final Logger LOGGER =
Logger.getLogger(RunnerStatusService.class.getName());
+
+ @Scheduled(every = "{karavan.runner-status-interval}", concurrentExecution
= Scheduled.ConcurrentExecution.SKIP)
+ void collectRunnerStatus() {
+ System.out.println("collectRunnerStatus");
+ dockerService.getRunnerContainer().forEach(container -> {
+ String name = container.getNames()[0].replace("/", "");
+ String projectId = name.replace("-" + Constants.RUNNER_SUFFIX, "");
+ PodStatus ps = getPodStatus(container.getId(), name, projectId,
container.getState(), container.getCreated());
+ infinispanService.savePodStatus(ps);
+ });
+ }
+
+ public PodStatus getPodStatus(String id, String name, String projectId,
String state, Long created) {
+ try {
+ boolean initialized = Arrays.asList("running",
"restarting").contains(state);
+ boolean ready = Arrays.asList("running",
"restarting").contains(state);
+ boolean terminating = Arrays.asList("paused",
"exited").contains(state);
+ String creationTimestamp = new Date(created).toString();
+
+ Statistics stats = dockerService.getContainerStats(id);
+
+ String requestMemory =
Objects.requireNonNull(stats.getMemoryStats().getUsage()).toString();
+ String requestCpu = "N/A";
+ String limitMemory =
Objects.requireNonNull(stats.getMemoryStats().getLimit()).toString();
+ String limitCpu = "N/A";
+ return new PodStatus(
+ name,
+ state,
+ initialized,
+ ready,
+ terminating,
+ "",
+ name,
+ projectId,
+ environment,
+ true,
+ requestMemory,
+ requestCpu,
+ limitMemory,
+ limitCpu,
+ creationTimestamp
+ );
+ } catch (Exception ex) {
+ LOGGER.error(ex.getMessage(), ex.getCause());
+ return new PodStatus(
+ name,
+ projectId,
+ environment);
+ }
+ }
+}
diff --git
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/docker/DockerEventListener.java
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/docker/DockerEventListener.java
index 9ea88796..2ca02b23 100644
---
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/docker/DockerEventListener.java
+++
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/docker/DockerEventListener.java
@@ -4,8 +4,13 @@ import com.github.dockerjava.api.async.ResultCallback;
import com.github.dockerjava.api.model.Container;
import com.github.dockerjava.api.model.Event;
import com.github.dockerjava.api.model.EventType;
-import org.apache.camel.karavan.bashi.HealthChecker;
-import org.apache.camel.karavan.bashi.KaravanContainers;
+import com.github.dockerjava.api.model.Statistics;
+import io.vertx.core.eventbus.EventBus;
+import org.apache.camel.karavan.bashi.ConductorService;
+import org.apache.camel.karavan.bashi.Constants;
+import org.apache.camel.karavan.bashi.infinispan.InfinispanService;
+import org.apache.camel.karavan.bashi.infinispan.PodStatus;
+import org.eclipse.microprofile.config.inject.ConfigProperty;
import org.jboss.logging.Logger;
import javax.enterprise.context.ApplicationScoped;
@@ -13,17 +18,24 @@ import javax.inject.Inject;
import java.io.Closeable;
import java.io.IOException;
import java.util.Arrays;
+import java.util.Date;
import java.util.Objects;
@ApplicationScoped
public class DockerEventListener implements ResultCallback<Event> {
- @Inject
- KaravanContainers karavanContainers;
+ @ConfigProperty(name = "karavan.environment")
+ String environment;
@Inject
DockerService dockerService;
+ @Inject
+ EventBus eventBus;
+
+ @Inject
+ InfinispanService infinispanService;
+
private static final Logger LOGGER =
Logger.getLogger(DockerEventListener.class.getName());
@Override
@@ -34,16 +46,21 @@ public class DockerEventListener implements
ResultCallback<Event> {
@Override
public void onNext(Event event) {
// LOGGER.info(event.getType() + " : " + event.getStatus());
- if (Objects.equals(event.getType(), EventType.CONTAINER)){
- Container c = dockerService.getContainer(event.getId());
- if (Arrays.asList("stop", "die", "kill", "pause",
"destroy").contains(event.getStatus())) {
- karavanContainers.removeContainer(c.getId());
- } else if (Arrays.asList("start",
"unpause").contains(event.getStatus())) {
- karavanContainers.addContainer(c, "unknown");
- } else if (event.getStatus().startsWith("health_status:")) {
- String health = event.getStatus().replace("health_status: ",
"");
- LOGGER.info(event.getType() + " : " + event.getId() + " : " +
health);
- karavanContainers.addContainer(c, health);
+ if (Objects.equals(event.getType(), EventType.CONTAINER)) {
+ Container container = dockerService.getContainer(event.getId());
+ String status = event.getStatus();
+ if (container.getNames()[0].equals("/infinispan") &&
status.startsWith("health_status:")) {
+ String health = status.replace("health_status: ", "");
+ LOGGER.infof("Container %s health status: %s",
container.getNames()[0], health);
+ eventBus.publish(ConductorService.ADDRESS_INFINISPAN_HEALTH,
health);
+ } else if
(container.getNames()[0].endsWith(Constants.RUNNER_SUFFIX)) {
+ if (Arrays.asList("stop", "die", "kill", "pause",
"destroy").contains(event.getStatus())) {
+ String name = container.getNames()[0].replace("/", "");
+ String projectId = name.replace("-" +
Constants.RUNNER_SUFFIX, "");
+ infinispanService.deletePodStatus(new PodStatus(name,
projectId, environment));
+ } else if (Arrays.asList("start",
"unpause").contains(event.getStatus())) {
+
+ }
}
}
}
@@ -62,4 +79,6 @@ public class DockerEventListener implements
ResultCallback<Event> {
public void close() throws IOException {
LOGGER.error("DockerEventListener close");
}
+
+
}
diff --git
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/docker/DockerService.java
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/docker/DockerService.java
index 89d54d74..44fa6824 100644
---
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/docker/DockerService.java
+++
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/docker/DockerService.java
@@ -9,21 +9,23 @@ import com.github.dockerjava.api.model.*;
import com.github.dockerjava.core.DefaultDockerClientConfig;
import com.github.dockerjava.core.DockerClientConfig;
import com.github.dockerjava.core.DockerClientImpl;
+import com.github.dockerjava.core.InvocationBuilder;
import com.github.dockerjava.httpclient5.ApacheDockerHttpClient;
import com.github.dockerjava.transport.DockerHttpClient;
-import org.apache.camel.karavan.bashi.HealthChecker;
-import org.apache.camel.karavan.bashi.KaravanContainers;
+import io.vertx.core.eventbus.EventBus;
import org.jboss.logging.Logger;
import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;
+import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
-import static org.apache.camel.karavan.bashi.KaravanConstants.NETWORK_NAME;
+import static
org.apache.camel.karavan.bashi.ConductorService.ADDRESS_INFINISPAN_HEALTH;
+import static org.apache.camel.karavan.bashi.Constants.NETWORK_NAME;
@ApplicationScoped
public class DockerService {
@@ -34,14 +36,15 @@ public class DockerService {
DockerEventListener dockerEventListener;
@Inject
- KaravanContainers karavanContainers;
+ EventBus eventBus;
public void startListeners() {
getDockerClient().eventsCmd().exec(dockerEventListener);
}
public void createNetwork() {
- if (!getDockerClient().listNetworksCmd().exec().stream().filter(n ->
n.getName().equals(NETWORK_NAME))
+ if (!getDockerClient().listNetworksCmd().exec().stream()
+ .filter(n -> n.getName().equals(NETWORK_NAME))
.findFirst().isPresent()) {
CreateNetworkResponse res =
getDockerClient().createNetworkCmd().withName(NETWORK_NAME).withAttachable(true).exec();
LOGGER.info("Network created: {}" + res);
@@ -55,7 +58,9 @@ public class DockerService {
.filter(c -> c.getState().equals("running"))
.forEach(c -> {
HealthState hs =
getDockerClient().inspectContainerCmd(c.getId()).exec().getState().getHealth();
- karavanContainers.addContainer(c, hs != null ?
hs.getStatus() : "unknown");
+ if (c.getNames()[0].equals("/infinispan")) {
+ eventBus.publish(ADDRESS_INFINISPAN_HEALTH,
hs.getStatus());
+ }
});
}
@@ -64,7 +69,26 @@ public class DockerService {
return containers.get(0);
}
- public void createContainer(String name, String image, List<String> env,
String ports, boolean exposedPort, HealthCheck healthCheck) throws
InterruptedException {
+ public List<Container> getRunnerContainer() {
+ return getDockerClient().listContainersCmd()
+ .withShowAll(true).withLabelFilter(Map.of("type",
"runner")).exec();
+ }
+
+ public Statistics getContainerStats(String id) {
+ InvocationBuilder.AsyncResultCallback<Statistics> callback = new
InvocationBuilder.AsyncResultCallback<>();
+ getDockerClient().statsCmd(id).withContainerId(id).exec(callback);
+ Statistics stats = null;
+ try {
+ stats = callback.awaitResult();
+ callback.close();
+ } catch (RuntimeException | IOException e) {
+ // you may want to throw an exception here
+ }
+ return stats;
+ }
+
+ public void createContainer(String name, String image, List<String> env,
String ports,
+ boolean exposedPort, HealthCheck healthCheck,
Map<String, String> labels) throws InterruptedException {
List<Container> containers =
getDockerClient().listContainersCmd().withShowAll(true).withNameFilter(List.of(name)).exec();
if (containers.size() == 0) {
pullImage(image);
@@ -73,10 +97,11 @@ public class DockerService {
CreateContainerResponse container =
getDockerClient().createContainerCmd(image)
.withName(name)
+ .withLabels(labels)
.withEnv(env)
.withExposedPorts(exposedPorts)
.withHostName(name)
- .withHostConfig(getHostConfig(ports))
+ .withHostConfig(getHostConfig(ports, exposedPort))
.withHealthcheck(healthCheck)
.exec();
LOGGER.info("Container created: " + container.getId());
@@ -100,7 +125,7 @@ public class DockerService {
startContainer(name);
}
- public void stopContainer(String name) throws InterruptedException {
+ public void stopContainer(String name) {
List<Container> containers =
getDockerClient().listContainersCmd().withShowAll(true).withNameFilter(List.of(name)).exec();
if (containers.size() == 1) {
Container container = containers.get(0);
@@ -110,29 +135,42 @@ public class DockerService {
}
}
+ public void deleteContainer(String name) {
+ List<Container> containers =
getDockerClient().listContainersCmd().withShowAll(true).withNameFilter(List.of(name)).exec();
+ if (containers.size() == 1) {
+ Container container = containers.get(0);
+ getDockerClient().removeContainerCmd(container.getId()).exec();
+ }
+ }
+
public void pullImage(String image) throws InterruptedException {
List<Image> images =
getDockerClient().listImagesCmd().withShowAll(true).exec();
if (!images.stream().filter(i ->
Arrays.asList(i.getRepoTags()).contains(image)).findFirst().isPresent()) {
- ResultCallback.Adapter<PullResponseItem> pull =
getDockerClient().pullImageCmd(image).start().awaitCompletion();
+ ResultCallback.Adapter<PullResponseItem> pull =
getDockerClient().pullImageCmd(image).start().awaitCompletion();
}
}
- private HostConfig getHostConfig(String ports) {
+ private HostConfig getHostConfig(String ports, boolean exposedPort) {
Ports portBindings = new Ports();
getPortsFromString(ports).forEach((hostPort, containerPort) -> {
- portBindings.bind(ExposedPort.tcp(containerPort),
Ports.Binding.bindIp("0.0.0.0").bindPort(hostPort));
+ portBindings.bind(
+ ExposedPort.tcp(containerPort),
+ exposedPort ?
Ports.Binding.bindIp("0.0.0.0").bindPort(hostPort) :
Ports.Binding.bindPort(hostPort)
+ );
});
return new HostConfig()
.withPortBindings(portBindings)
.withNetworkMode(NETWORK_NAME);
}
- private Map<Integer,Integer> getPortsFromString(String ports){
- Map<Integer,Integer> p = new HashMap<>();
- Arrays.stream(ports.split(",")).forEach(s -> {
- String[] values = s.split(":");
- p.put(Integer.parseInt(values[0]), Integer.parseInt(values[1]));
- });
+ private Map<Integer, Integer> getPortsFromString(String ports) {
+ Map<Integer, Integer> p = new HashMap<>();
+ if (!ports.isEmpty()) {
+ Arrays.stream(ports.split(",")).forEach(s -> {
+ String[] values = s.split(":");
+ p.put(Integer.parseInt(values[0]),
Integer.parseInt(values[1]));
+ });
+ }
return p;
}
diff --git
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/infinispan/ClientRunnerListener.java
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/infinispan/ClientRunnerListener.java
new file mode 100644
index 00000000..1eae6fe3
--- /dev/null
+++
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/infinispan/ClientRunnerListener.java
@@ -0,0 +1,49 @@
+package org.apache.camel.karavan.bashi.infinispan;
+
+import io.vertx.core.eventbus.EventBus;
+import io.vertx.core.json.JsonObject;
+import org.apache.camel.karavan.bashi.ConductorService;
+import org.infinispan.client.hotrod.annotation.ClientCacheEntryCreated;
+import org.infinispan.client.hotrod.annotation.ClientCacheEntryModified;
+import org.infinispan.client.hotrod.annotation.ClientListener;
+import org.infinispan.client.hotrod.event.ClientCacheEntryCreatedEvent;
+import org.infinispan.client.hotrod.event.ClientCacheEntryModifiedEvent;
+
+import java.util.Objects;
+
+@ClientListener
+public class ClientRunnerListener {
+
+ private final InfinispanService infinispanService;
+ private final EventBus eventBus;
+
+ public ClientRunnerListener(InfinispanService infinispanService, EventBus
eventBus) {
+ this.infinispanService = infinispanService;
+ this.eventBus = eventBus;
+ }
+
+ @ClientCacheEntryCreated
+ public void entryCreated(ClientCacheEntryCreatedEvent<GroupedKey> event) {
+ System.out.println("entryCreated");
+ String command = event.getKey().getKey();
+ String projectId = event.getKey().getGroup();
+ if (Objects.equals(command, RunnerCommand.NAME.run.name())) {
+ eventBus.publish(ConductorService.ADDRESS_RUNNER,
JsonObject.of("projectId", projectId, "command", command, "isRunner", true));
+ } else if (Objects.equals(command, RunnerCommand.NAME.delete.name())) {
+ eventBus.publish(ConductorService.ADDRESS_RUNNER,
JsonObject.of("projectId", projectId, "command", command, "isRunner", true));
+ }
+ }
+
+ @ClientCacheEntryModified
+ public void entryModified(ClientCacheEntryModifiedEvent<GroupedKey> event)
{
+ System.out.println("entryCreated");
+ String command = event.getKey().getKey();
+ String projectId = event.getKey().getGroup();
+ if (Objects.equals(command, RunnerCommand.NAME.run.name())) {
+ eventBus.publish(ConductorService.ADDRESS_RUNNER,
JsonObject.of("projectId", projectId, "command", command, "isRunner", true));
+ } else if (Objects.equals(command, RunnerCommand.NAME.delete.name())) {
+ eventBus.publish(ConductorService.ADDRESS_RUNNER,
JsonObject.of("projectId", projectId, "command", command, "isRunner", true));
+ }
+ }
+
+}
\ No newline at end of file
diff --git
a/karavan-app/src/main/java/org/apache/camel/karavan/model/GroupedKey.java
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/infinispan/GroupedKey.java
similarity index 79%
copy from
karavan-app/src/main/java/org/apache/camel/karavan/model/GroupedKey.java
copy to
karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/infinispan/GroupedKey.java
index a50e5ced..550a1d54 100644
--- a/karavan-app/src/main/java/org/apache/camel/karavan/model/GroupedKey.java
+++
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/infinispan/GroupedKey.java
@@ -1,4 +1,4 @@
-package org.apache.camel.karavan.model;
+package org.apache.camel.karavan.bashi.infinispan;
import org.infinispan.protostream.annotations.ProtoFactory;
import org.infinispan.protostream.annotations.ProtoField;
@@ -22,6 +22,18 @@ public class GroupedKey {
}
+ public void setGroup(String group) {
+ this.group = group;
+ }
+
+ public String getKey() {
+ return key;
+ }
+
+ public void setKey(String key) {
+ this.key = key;
+ }
+
public String getGroup() {
return group;
}
diff --git
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/infinispan/InfinispanService.java
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/infinispan/InfinispanService.java
new file mode 100644
index 00000000..a2223ae1
--- /dev/null
+++
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/infinispan/InfinispanService.java
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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
+ *
+ * http://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 org.apache.camel.karavan.bashi.infinispan;
+
+import io.quarkus.vertx.ConsumeEvent;
+import io.vertx.core.eventbus.EventBus;
+import org.eclipse.microprofile.config.inject.ConfigProperty;
+import org.infinispan.client.hotrod.RemoteCache;
+import org.infinispan.client.hotrod.RemoteCacheManager;
+import org.infinispan.client.hotrod.Search;
+import org.infinispan.client.hotrod.configuration.ClientIntelligence;
+import org.infinispan.client.hotrod.configuration.ConfigurationBuilder;
+import org.infinispan.commons.api.BasicCache;
+import org.infinispan.commons.configuration.StringConfiguration;
+import org.infinispan.commons.marshall.ProtoStreamMarshaller;
+import org.infinispan.query.dsl.QueryFactory;
+import org.jboss.logging.Logger;
+
+import javax.enterprise.context.ApplicationScoped;
+import javax.enterprise.inject.Default;
+import javax.inject.Inject;
+
+import java.util.List;
+
+import static
org.apache.camel.karavan.bashi.ConductorService.ADDRESS_INFINISPAN_HEALTH;
+
+@Default
+@ApplicationScoped
+public class InfinispanService {
+
+ @ConfigProperty(name ="quarkus.infinispan-client.hosts")
+ String infinispanHosts;
+ @ConfigProperty(name ="quarkus.infinispan-client.username")
+ String infinispanUsername;
+ @ConfigProperty(name ="quarkus.infinispan-client.password")
+ String infinispanPassword;
+
+ @Inject
+ EventBus eventBus;
+
+ private BasicCache<GroupedKey, PodStatus> podStatuses;
+ private BasicCache<GroupedKey, String> runnerCommands;
+
+ private static final String CACHE_CONFIG = "<distributed-cache
name=\"%s\">"
+ + " <encoding media-type=\"application/x-protostream\"/>"
+ + " <groups enabled=\"true\"/>"
+ + "</distributed-cache>";
+
+ private static final Logger LOGGER =
Logger.getLogger(InfinispanService.class.getName());
+
+ @ConsumeEvent(value = ADDRESS_INFINISPAN_HEALTH, blocking = true, ordered
= true)
+ void startService(String infinispanHealth) {
+ if (infinispanHealth.equals("healthy")) {
+ LOGGER.info("InfinispanService is starting in remote mode");
+
+ ProtoStreamMarshaller marshaller = new ProtoStreamMarshaller();
+ marshaller.register(new ProjectStoreSchemaImpl());
+
+ ConfigurationBuilder builder = new ConfigurationBuilder();
+ builder.socketTimeout(1000)
+ .connectionTimeout(10000)
+ .addServers(infinispanHosts)
+ .security()
+ .authentication().enable()
+ .username(infinispanUsername)
+ .password(infinispanPassword)
+ .clientIntelligence(ClientIntelligence.BASIC)
+ .marshaller(marshaller);
+
+ RemoteCacheManager cacheManager = new
RemoteCacheManager(builder.build());
+ runnerCommands =
cacheManager.administration().getOrCreateCache(RunnerCommand.CACHE, new
StringConfiguration(String.format(CACHE_CONFIG, RunnerCommand.CACHE)));
+ podStatuses =
cacheManager.administration().getOrCreateCache(PodStatus.CACHE, new
StringConfiguration(String.format(CACHE_CONFIG, PodStatus.CACHE)));
+ cacheManager.getCache(RunnerCommand.CACHE).addClientListener(new
ClientRunnerListener(this, eventBus));
+ }
+ }
+
+ public void sendRunnerCommand(String projectId, String runnerName,
RunnerCommand.NAME command) {
+ runnerCommands.put(GroupedKey.create(projectId, runnerName),
command.name());
+ }
+
+ public String getRunnerCommand(GroupedKey key) {
+ return runnerCommands.get(key);
+ }
+
+ public List<PodStatus> getPodStatuses(String projectId, String env) {
+ QueryFactory queryFactory = Search.getQueryFactory((RemoteCache<?, ?>)
podStatuses);
+ return queryFactory.<PodStatus>create("FROM karavan.PodStatus WHERE
project = :project AND env = :env")
+ .setParameter("project", projectId)
+ .setParameter("env", env)
+ .execute().list();
+ }
+
+ public List<PodStatus> getPodStatuses(String env) {
+ QueryFactory queryFactory = Search.getQueryFactory((RemoteCache<?, ?>)
podStatuses);
+ return queryFactory.<PodStatus>create("FROM karavan.PodStatus WHERE
env = :env")
+ .setParameter("env", env)
+ .execute().list();
+ }
+
+ public void savePodStatus(PodStatus status) {
+ podStatuses.put(GroupedKey.create(status.getProject(),
status.getName()), status);
+ }
+
+ public void deletePodStatus(PodStatus status) {
+ podStatuses.remove(GroupedKey.create(status.getProject(),
status.getName()));
+ }
+
+}
diff --git
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/infinispan/PodStatus.java
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/infinispan/PodStatus.java
new file mode 100644
index 00000000..31ee43a8
--- /dev/null
+++
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/infinispan/PodStatus.java
@@ -0,0 +1,209 @@
+package org.apache.camel.karavan.bashi.infinispan;
+
+import org.infinispan.protostream.annotations.ProtoFactory;
+import org.infinispan.protostream.annotations.ProtoField;
+
+public class PodStatus {
+ public static final String CACHE = "pod_statuses";
+ @ProtoField(number = 1)
+ String name;
+ @ProtoField(number = 2)
+ String phase;
+ @ProtoField(number = 3)
+ Boolean initialized;
+ @ProtoField(number = 4)
+ Boolean ready;
+ @ProtoField(number = 5)
+ Boolean terminating;
+ @ProtoField(number = 6)
+ String reason;
+ @ProtoField(number = 7)
+ String deployment;
+ @ProtoField(number = 8)
+ String project;
+ @ProtoField(number = 9)
+ String env;
+ @ProtoField(number = 10)
+ Boolean runner;
+ @ProtoField(number = 11)
+ String requestMemory;
+ @ProtoField(number = 12)
+ String requestCpu;
+ @ProtoField(number = 13)
+ String limitMemory;
+ @ProtoField(number = 14)
+ String limitCpu;
+ @ProtoField(number = 15)
+ String creationTimestamp;
+
+ public PodStatus(String name, String project, String env) {
+ this.name = name;
+ this.phase = "";
+ this.initialized = false;
+ this.ready = false;
+ this.terminating = false;
+ this.reason = "";
+ this.project = project;
+ this.env = env;
+ }
+
+ @ProtoFactory
+ public PodStatus(String name, String phase, Boolean initialized, Boolean
ready, Boolean terminating, String reason, String deployment, String project,
String env, Boolean runner, String requestMemory, String requestCpu, String
limitMemory, String limitCpu, String creationTimestamp) {
+ this.name = name;
+ this.phase = phase;
+ this.initialized = initialized;
+ this.ready = ready;
+ this.terminating = terminating;
+ this.reason = reason;
+ this.deployment = deployment;
+ this.project = project;
+ this.env = env;
+ this.runner = runner;
+ this.requestMemory = requestMemory;
+ this.requestCpu = requestCpu;
+ this.limitMemory = limitMemory;
+ this.limitCpu = limitCpu;
+ this.creationTimestamp = creationTimestamp;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public String getPhase() {
+ return phase;
+ }
+
+ public void setPhase(String phase) {
+ this.phase = phase;
+ }
+
+ public Boolean getInitialized() {
+ return initialized;
+ }
+
+ public void setInitialized(Boolean initialized) {
+ this.initialized = initialized;
+ }
+
+ public Boolean getReady() {
+ return ready;
+ }
+
+ public void setReady(Boolean ready) {
+ this.ready = ready;
+ }
+
+ public Boolean getTerminating() {
+ return terminating;
+ }
+
+ public void setTerminating(Boolean terminating) {
+ this.terminating = terminating;
+ }
+
+ public String getReason() {
+ return reason;
+ }
+
+ public void setReason(String reason) {
+ this.reason = reason;
+ }
+
+ public String getDeployment() {
+ return deployment;
+ }
+
+ public void setDeployment(String deployment) {
+ this.deployment = deployment;
+ }
+
+ public String getProject() {
+ return project;
+ }
+
+ public void setProject(String project) {
+ this.project = project;
+ }
+
+ public String getEnv() {
+ return env;
+ }
+
+ public void setEnv(String env) {
+ this.env = env;
+ }
+
+ public Boolean getRunner() {
+ return runner;
+ }
+
+ public void setRunner(Boolean runner) {
+ this.runner = runner;
+ }
+
+ public String getRequestMemory() {
+ return requestMemory;
+ }
+
+ public void setRequestMemory(String requestMemory) {
+ this.requestMemory = requestMemory;
+ }
+
+ public String getRequestCpu() {
+ return requestCpu;
+ }
+
+ public void setRequestCpu(String requestCpu) {
+ this.requestCpu = requestCpu;
+ }
+
+ public String getLimitMemory() {
+ return limitMemory;
+ }
+
+ public void setLimitMemory(String limitMemory) {
+ this.limitMemory = limitMemory;
+ }
+
+ public String getLimitCpu() {
+ return limitCpu;
+ }
+
+ public void setLimitCpu(String limitCpu) {
+ this.limitCpu = limitCpu;
+ }
+
+ public String getCreationTimestamp() {
+ return creationTimestamp;
+ }
+
+ public void setCreationTimestamp(String creationTimestamp) {
+ this.creationTimestamp = creationTimestamp;
+ }
+
+ @Override
+ public String toString() {
+ return "PodStatus{" +
+ "name='" + name + '\'' +
+ ", phase='" + phase + '\'' +
+ ", initialized=" + initialized +
+ ", ready=" + ready +
+ ", terminating=" + terminating +
+ ", reason='" + reason + '\'' +
+ ", deployment='" + deployment + '\'' +
+ ", project='" + project + '\'' +
+ ", env='" + env + '\'' +
+ ", runner=" + runner +
+ ", requestMemory='" + requestMemory + '\'' +
+ ", requestCpu='" + requestCpu + '\'' +
+ ", limitMemory='" + limitMemory + '\'' +
+ ", limitCpu='" + limitCpu + '\'' +
+ ", creationTimestamp='" + creationTimestamp + '\'' +
+ '}';
+ }
+}
diff --git
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/infinispan/ProjectStoreSchema.java
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/infinispan/ProjectStoreSchema.java
new file mode 100644
index 00000000..bf3ea015
--- /dev/null
+++
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/infinispan/ProjectStoreSchema.java
@@ -0,0 +1,8 @@
+package org.apache.camel.karavan.bashi.infinispan;
+
+import org.infinispan.protostream.GeneratedSchema;
+import org.infinispan.protostream.annotations.AutoProtoSchemaBuilder;
+
+@AutoProtoSchemaBuilder(includeClasses = {GroupedKey.class, PodStatus.class},
schemaPackageName = "karavan")
+public interface ProjectStoreSchema extends GeneratedSchema {
+}
diff --git
a/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/infinispan/RunnerCommand.java
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/infinispan/RunnerCommand.java
new file mode 100644
index 00000000..1514920d
--- /dev/null
+++
b/karavan-bashi/src/main/java/org/apache/camel/karavan/bashi/infinispan/RunnerCommand.java
@@ -0,0 +1,13 @@
+package org.apache.camel.karavan.bashi.infinispan;
+
+public class RunnerCommand {
+
+ public enum NAME {
+ run,
+ delete,
+ reload
+ }
+
+ public static final String CACHE = "runner_commands";
+
+}
diff --git a/karavan-bashi/src/main/resources/application.properties
b/karavan-bashi/src/main/resources/application.properties
index 8649065e..932ff649 100644
--- a/karavan-bashi/src/main/resources/application.properties
+++ b/karavan-bashi/src/main/resources/application.properties
@@ -1,16 +1,26 @@
+# Infinispan container config in Docker
infinispan.image=quay.io/infinispan/server:14.0.6.Final
infinispan.port=11222:11222
infinispan.username=admin
infinispan.password=karavan
+# Karavan container config in Docker
karavan.image=marat/karavan:3.21.1-SNAPSHOT
+karavan.runner-image=ghcr.io/apache/camel-karavan-runner:3.21.0
karavan.port=8080:8080
karavan.environment=dev
karavan.default-runtime=quarkus
karavan.runtimes=quarkus,spring-boot
+karavan.runner-status-interval=2s
# Git repository Configuration
karavan.git-repository=${GIT_REPOSITORY}
karavan.git-username=${GIT_USERNAME}
karavan.git-password=${GIT_TOKEN}
karavan.git-branch=main
+
+# Infinispan Client configuration
+quarkus.infinispan-client.hosts=localhost:11222
+quarkus.infinispan-client.username=admin
+quarkus.infinispan-client.password=karavan
+quarkus.infinispan-client.devservices.enabled=false
\ No newline at end of file
