-
Notifications
You must be signed in to change notification settings - Fork 17
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: add job verticle to deploy missing entity verticles
- Loading branch information
Showing
4 changed files
with
367 additions
and
7 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,195 @@ | ||
| package io.neonbee.job; | ||
|
|
||
| import static io.neonbee.internal.deploy.DeployableVerticle.fromClass; | ||
| import static io.neonbee.internal.deploy.Deployables.fromDeployables; | ||
| import static io.neonbee.internal.scanner.DeployableScanner.scanForDeployableClasses; | ||
|
|
||
| import java.time.Duration; | ||
| import java.util.AbstractMap; | ||
| import java.util.Collection; | ||
| import java.util.HashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Set; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| import com.google.common.annotations.VisibleForTesting; | ||
|
|
||
| import io.neonbee.NeonBee; | ||
| import io.neonbee.NeonBeeDeployable; | ||
| import io.neonbee.NeonBeeProfile; | ||
| import io.neonbee.data.DataContext; | ||
| import io.neonbee.entity.EntityVerticle; | ||
| import io.neonbee.internal.Registry; | ||
| import io.neonbee.internal.cluster.ClusterHelper; | ||
| import io.neonbee.internal.cluster.entity.ClusterEntityRegistry; | ||
| import io.neonbee.internal.deploy.Deployable; | ||
| import io.neonbee.internal.deploy.Deployables; | ||
| import io.neonbee.logging.LoggingFacade; | ||
| import io.vertx.core.Future; | ||
| import io.vertx.core.Verticle; | ||
| import io.vertx.core.Vertx; | ||
| import io.vertx.core.json.JsonArray; | ||
| import io.vertx.core.json.JsonObject; | ||
|
|
||
| /** | ||
| * A job that redeploys all entity verticles that are not deployed in the cluster. | ||
| */ | ||
| @NeonBeeDeployable(namespace = NeonBeeDeployable.NEONBEE_NAMESPACE, autoDeploy = false) | ||
| public class RedeployEntitiesJob extends JobVerticle { | ||
|
|
||
| private static final LoggingFacade LOGGER = LoggingFacade.create(RedeployEntitiesJob.class); | ||
|
|
||
| private static final Duration DEFAULT_INTERVAL = Duration.ofMinutes(5L); | ||
|
|
||
| /** | ||
| * Create a new ReregisterEntitiesJob job verticle with the default configuration. | ||
| */ | ||
| public RedeployEntitiesJob() { | ||
| this(new JobSchedule(DEFAULT_INTERVAL)); | ||
| } | ||
|
|
||
| /** | ||
| * Create a new ReregisterEntitiesJob job verticle. | ||
| * | ||
| * @param schedule the schedule to use when starting this job verticle | ||
| */ | ||
| public RedeployEntitiesJob(JobSchedule schedule) { | ||
| this(schedule, false); | ||
| } | ||
|
|
||
| /** | ||
| * Create a new ReregisterEntitiesJob job verticle. Optionally undeploy the verticle when the job execution ended | ||
| * (hit the end instant or one time execution) | ||
| * | ||
| * @param schedule the schedule to use when starting this job verticle | ||
| * @param undeployWhenDone if true, undeploy the verticle when done | ||
| */ | ||
| public RedeployEntitiesJob(JobSchedule schedule, boolean undeployWhenDone) { | ||
| super(schedule, undeployWhenDone); | ||
| } | ||
|
|
||
| @Override | ||
| public Future<?> execute(DataContext context) { | ||
| LOGGER.correlateWith(context).info("Start re registering entities"); | ||
| long startTime = System.currentTimeMillis(); | ||
|
|
||
| // get the currently deployed entity verticles | ||
| NeonBee neonBee = NeonBee.get(getVertx()); | ||
| Registry<String> entityRegistry = neonBee.getEntityRegistry(); | ||
| if (entityRegistry instanceof ClusterEntityRegistry) { | ||
| LOGGER.correlateWith(context).info("Getting registered entities from cluster"); | ||
|
|
||
| ClusterEntityRegistry clusterEntityRegistry = ((ClusterEntityRegistry) entityRegistry); | ||
| Future<JsonArray> clusteringInformation = clusterEntityRegistry | ||
| .getClusteringInformation(ClusterHelper.getClusterNodeId(vertx)) | ||
| .onSuccess(event -> LOGGER.correlateWith(context).info("Got registered entities from cluster")) | ||
| .onFailure(error -> LOGGER.correlateWith(context) | ||
| .error("Failed getting registered entities from cluster", error)); | ||
|
|
||
| var entitiesFromClassPath = classPathEntityVertilces(vertx) | ||
| .onSuccess(event -> LOGGER.correlateWith(context) | ||
| .info("Finished reregistering entities. took {} ms", | ||
| System.currentTimeMillis() - startTime)) | ||
| .onFailure(error -> LOGGER.correlateWith(context) | ||
| .error("Failed reregistering entities", error)); | ||
|
|
||
| return Future.all(clusteringInformation, entitiesFromClassPath) | ||
| .map(compositeFuture -> findNotDeployedEntityVerticles( | ||
| context, | ||
| entitiesFromClassPath.result(), | ||
| clusteringInformation.result())) | ||
| .compose(difference -> deployNotDeployedEntityVerticles(context, neonBee, difference)) | ||
| .onFailure(error -> LOGGER.correlateWith(context) | ||
| .error("Failed getting registered entities from cluster", error)); | ||
| } else { | ||
| // if it is not a clustered deployment we have nothing to do | ||
| return Future.succeededFuture(); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Find all entity verticles that are not deployed in the cluster. | ||
| * | ||
| * @param context the data context | ||
| * @param classPathEntitiesMap the entity verticles from the class path | ||
| * @param clusteringInformation the clustering information | ||
| * @return a map of entity verticles that are not deployed in the cluster | ||
| */ | ||
| private Map<String, Class<? extends EntityVerticle>> findNotDeployedEntityVerticles( | ||
| DataContext context, | ||
| Map<String, Class<? extends EntityVerticle>> classPathEntitiesMap, | ||
| JsonArray clusteringInformation) { | ||
| Set<String> deployedEntitiesSet = qualifiedNamesSet(context, clusteringInformation); | ||
| Map<String, Class<? extends EntityVerticle>> difference = new HashMap<>(classPathEntitiesMap); | ||
| difference.keySet().removeAll(deployedEntitiesSet); | ||
| return difference; | ||
| } | ||
|
|
||
| private Set<String> qualifiedNamesSet(DataContext context, JsonArray clusteringInformation) { | ||
| Set<String> deployedEntitiesSet; | ||
| if (clusteringInformation == null) { | ||
| LOGGER.correlateWith(context).info("No entities registered in cluster"); | ||
| deployedEntitiesSet = Set.of(); | ||
| } else { | ||
| deployedEntitiesSet = clusteringInformation | ||
| .stream() | ||
| .map(jo -> (JsonObject) jo) | ||
| .map(jo -> jo.getString(ClusterEntityRegistry.QUALIFIED_NAME_KEY)) | ||
| .collect(Collectors.toSet()); | ||
| } | ||
| return deployedEntitiesSet; | ||
| } | ||
|
|
||
| private Future<Object> deployNotDeployedEntityVerticles(DataContext context, NeonBee neonBee, | ||
| Map<String, Class<? extends EntityVerticle>> difference) { | ||
| if (difference.isEmpty()) { | ||
| LOGGER.correlateWith(context).info("Everything OK, all EntityVerticles are deployed."); | ||
| return Future.succeededFuture(); | ||
| } else { | ||
| List<Future<? extends Deployable>> toDeplyoe = difference.entrySet() | ||
| .stream() | ||
| .peek(stringClassEntry -> LOGGER.correlateWith(context).warn( | ||
| "Entity verticles with qualified name \"{}\" will be deployed.", | ||
| stringClassEntry.getKey())) | ||
| .map(stringClassEntry -> fromClass(vertx, stringClassEntry.getValue())) | ||
| .collect(Collectors.toList()); | ||
|
|
||
| return fromDeployables(toDeplyoe) | ||
| .compose(Deployables.allTo(neonBee)) | ||
| .onSuccess(deployment -> LOGGER.correlateWith(context) | ||
| .info("Entity verticles deployed. \"{}\"", deployment.getDeployable().getIdentifier())) | ||
| .mapEmpty(); | ||
| } | ||
| } | ||
|
|
||
| private Future<Map<String, Class<? extends EntityVerticle>>> classPathEntityVertilces(Vertx vertx) { | ||
| return scanForDeployableClasses(vertx).map(verticles -> verticles.stream() | ||
| .filter(EntityVerticle.class::isAssignableFrom) | ||
| .filter(verticleClass -> filterByAutoDeployAndProfiles(verticleClass, activeProfiles())) | ||
| .map(verticleClass -> (Class<? extends EntityVerticle>) verticleClass) | ||
| .map(verticleClass -> { | ||
| NeonBeeDeployable annotation = verticleClass.getAnnotation(NeonBeeDeployable.class); | ||
| String namespace = annotation != null ? annotation.namespace() + "/" : ""; | ||
|
|
||
| return new AbstractMap.SimpleEntry<String, Class<? extends EntityVerticle>>( | ||
| namespace + EntityVerticle.getName(verticleClass), | ||
| verticleClass); | ||
| }) | ||
| .collect(Collectors.toMap( | ||
| Map.Entry::getKey, | ||
| Map.Entry::getValue, | ||
| (existingValue, newValue) -> existingValue))); | ||
| } | ||
|
|
||
| private Set<NeonBeeProfile> activeProfiles() { | ||
| return NeonBee.get(getVertx()).getOptions().getActiveProfiles(); | ||
| } | ||
|
|
||
| @VisibleForTesting | ||
| boolean filterByAutoDeployAndProfiles(Class<? extends Verticle> verticleClass, | ||
| Collection<NeonBeeProfile> activeProfiles) { | ||
| NeonBeeDeployable annotation = verticleClass.getAnnotation(NeonBeeDeployable.class); | ||
| return annotation.autoDeploy() && annotation.profile().isActive(activeProfiles); | ||
| } | ||
| } |
Oops, something went wrong.