Merge remote-tracking branch 'origin/master' into authz-2

.gitignore

@@ -38,3 +38,4 @@ target/
 scratch
 
 scripts/benchmark/infra/
+nohup.out

buildSrc/src/main/groovy/com.flipkart.varadhi.java-common-conventions.gradle

@@ -17,7 +17,7 @@ repositories {
 ext {
  lombok_version = "1.18.26"
  slf4j_version = "2.0.7"
- vertx_version = "4.4.4"
+ vertx_version = "4.5.2"
 
  otl_version = "1.25.0"
  micrometer_version = "1.10.6"
@@ -67,6 +67,7 @@ dependencies {
  implementation("io.vertx:vertx-web:$vertx_version")
  implementation("io.vertx:vertx-auth-common:$vertx_version")
  implementation("io.vertx:vertx-auth-jwt:$vertx_version")
+ implementation("io.vertx:vertx-zookeeper:$vertx_version")
  implementation("io.vertx:vertx-opentelemetry:$vertx_version")
  implementation("io.vertx:vertx-opentracing:$vertx_version")
  implementation("io.vertx:vertx-micrometer-metrics:$vertx_version")

server/build.gradle

@@ -27,6 +27,7 @@ dependencies {
  implementation("io.vertx:vertx-web")
  implementation("io.vertx:vertx-auth-common")
  implementation("io.vertx:vertx-auth-jwt")
+ implementation("io.vertx:vertx-zookeeper")
  implementation("io.vertx:vertx-opentelemetry")
  implementation("io.vertx:vertx-opentracing")
  implementation("io.vertx:vertx-micrometer-metrics")

server/src/main/java/com/flipkart/varadhi/CoreServices.java

@@ -1,7 +1,7 @@
 package com.flipkart.varadhi;
 
 
-import com.flipkart.varadhi.config.ServerConfiguration;
+import com.flipkart.varadhi.config.ServerConfig;
 import com.flipkart.varadhi.spi.db.MetaStoreOptions;
 import com.flipkart.varadhi.spi.db.MetaStoreProvider;
 import com.flipkart.varadhi.spi.services.MessagingStackOptions;
@@ -39,7 +39,7 @@ public class CoreServices {
  private final MessagingStackProvider messagingStackProvider;
  private final MetaStoreProvider metaStoreProvider;
 
- public CoreServices(ServerConfiguration configuration) {
+ public CoreServices(ServerConfig configuration) {
  this.observabilityStack = setupObservabilityStack(configuration);
  this.messagingStackProvider = setupMessagingStackProvider(configuration.getMessagingStackOptions());
  this.metaStoreProvider = setupMetaStoreProvider(configuration.getMetaStoreOptions());
@@ -76,7 +76,7 @@ private MessagingStackProvider setupMessagingStackProvider(MessagingStackOptions
  return provider;
  }
 
- private ObservabilityStack setupObservabilityStack(ServerConfiguration configuration) {
+ private ObservabilityStack setupObservabilityStack(ServerConfig configuration) {
  Resource resource = Resource.getDefault()
  .merge(Resource.create(Attributes.of(ResourceAttributes.SERVICE_NAME, "com.flipkart.varadhi")));
 

server/src/main/java/com/flipkart/varadhi/Server.java

@@ -1,6 +1,7 @@
 package com.flipkart.varadhi;
 
-import com.flipkart.varadhi.config.ServerConfiguration;
+import com.flipkart.varadhi.cluster.custom.ZookeeperClusterManager;
+import com.flipkart.varadhi.config.ServerConfig;
 import com.flipkart.varadhi.deployment.FullDeploymentVerticleDeployer;
 import com.flipkart.varadhi.deployment.LeanDeploymentVerticleDeployer;
 import com.flipkart.varadhi.exceptions.InvalidConfigException;
@@ -16,6 +17,8 @@
 import io.vertx.tracing.opentelemetry.OpenTelemetryOptions;
 import lombok.extern.slf4j.Slf4j;
 
+import java.util.concurrent.ExecutionException;
+
 @Slf4j
 public class Server {
 
@@ -24,9 +27,9 @@ public static void main(String[] args) {
  try {
  String hostName = HostUtils.getHostName();
  log.info("Server Starting on {}.", hostName);
- ServerConfiguration configuration = readConfiguration(args);
+ ServerConfig configuration = readConfiguration(args);
  CoreServices services = new CoreServices(configuration);
- Vertx vertx = createVertex(configuration, services);
+ Vertx vertx = createVertx(configuration, services);
  deployVerticle(hostName, configuration, services, vertx);
  log.info("Server Started on {}.", hostName);
  } catch (Exception e) {
@@ -36,22 +39,33 @@ public static void main(String[] args) {
  // TODO: check need for shutdown hook
  }
 
- private static Vertx createVertex(ServerConfiguration configuration, CoreServices services) {
+ private static Vertx createVertx(ServerConfig configuration, CoreServices services)
+ throws ExecutionException, InterruptedException {
  log.debug("Creating Vertex");
+
  VertxOptions vertxOptions = configuration.getVertxOptions()
  .setTracingOptions(new OpenTelemetryOptions(services.getOpenTelemetry()))
  .setMetricsOptions(new MicrometerMetricsOptions()
  .setMicrometerRegistry(services.getMetricsRegistry())
  .setRegistryName("default")
  .setJvmMetricsEnabled(true)
  .setEnabled(true));
- Vertx vertx = Vertx.vertx(vertxOptions);
+ ZookeeperClusterManager clusterManager = new ZookeeperClusterManager(
+ configuration.getZookeeperOptions(),
+ configuration.getNodeId(),
+ configuration.getNodeResourcesOverride()
+ );
+ Vertx vertx = Vertx.builder()
+ .with(vertxOptions)
+ .withClusterManager(clusterManager)
+ .buildClustered()
+ .toCompletionStage().toCompletableFuture().get();
  log.debug("Created Vertex");
  return vertx;
  }
 
  private static void deployVerticle(
- String hostName, ServerConfiguration configuration, CoreServices services, Vertx vertx
+ String hostName, ServerConfig configuration, CoreServices services, Vertx vertx
  ) {
  log.debug("Verticle deployment started.");
  Tracer tracer = services.getTracer("varadhi");
@@ -84,15 +98,15 @@ private static void deployVerticle(
  }
 
 
- public static ServerConfiguration readConfiguration(String[] args) {
+ public static ServerConfig readConfiguration(String[] args) {
  if (args.length < 1) {
  log.error("Usage: java com.flipkart.varadhi.Server configuration.yml");
  System.exit(-1);
  }
  return readConfigFromFile(args[0]);
  }
 
- public static ServerConfiguration readConfigFromFile(String filePath) throws InvalidConfigException {
+ public static ServerConfig readConfigFromFile(String filePath) throws InvalidConfigException {
  log.info("Loading Configuration.");
  Vertx vertx = Vertx.vertx();
 
@@ -107,7 +121,7 @@ public static ServerConfiguration readConfigFromFile(String filePath) throws Inv
 
  try {
  JsonObject content = retriever.getConfig().toCompletionStage().toCompletableFuture().join();
- return content.mapTo(ServerConfiguration.class);
+ return content.mapTo(ServerConfig.class);
  } catch (Exception e) {
  throw new InvalidConfigException("Failed to load Application Configuration", e);
  } finally {

server/src/main/java/com/flipkart/varadhi/VerticleDeployer.java

@@ -2,7 +2,7 @@
 
 import com.flipkart.varadhi.auth.DefaultAuthorizationProvider;
 import com.flipkart.varadhi.config.RestOptions;
-import com.flipkart.varadhi.config.ServerConfiguration;
+import com.flipkart.varadhi.config.ServerConfig;
 import com.flipkart.varadhi.core.VaradhiTopicFactory;
 import com.flipkart.varadhi.core.VaradhiTopicService;
 import com.flipkart.varadhi.exceptions.VaradhiException;
@@ -51,7 +51,7 @@ public abstract class VerticleDeployer {
  public VerticleDeployer(
  String hostName,
  Vertx vertx,
- ServerConfiguration configuration,
+ ServerConfig configuration,
  MessagingStackProvider messagingStackProvider,
  MetaStoreProvider metaStoreProvider,
  MeterRegistry meterRegistry,
@@ -130,7 +130,7 @@ public List<RouteDefinition> getRouteDefinitions() {
  .toList();
  }
 
- public void deployVerticle(Vertx vertx, ServerConfiguration configuration) {
+ public void deployVerticle(Vertx vertx, ServerConfig configuration) {
  List<RouteDefinition> handlerDefinitions = getRouteDefinitions();
  if (shouldEnableAuthZHandlers(configuration)) {
  handlerDefinitions.addAll(iamPolicyHandlersSupplier.get().get());
@@ -151,10 +151,9 @@ public void deployVerticle(Vertx vertx, ServerConfiguration configuration) {
  .onSuccess(name -> log.debug("Successfully deployed the Verticle id({}).", name));
  }
 
- private boolean shouldEnableAuthZHandlers(ServerConfiguration configuration) {
+ private boolean shouldEnableAuthZHandlers(ServerConfig configuration) {
  String defaultProviderClass = DefaultAuthorizationProvider.class.getName();
  return configuration.isAuthorizationEnabled()
  && defaultProviderClass.equals(configuration.getAuthorization().getProviderClassName());
  }
-
 }

server/src/main/java/com/flipkart/varadhi/auth/DefaultAuthorizationProvider.java

@@ -2,7 +2,7 @@
 
 import com.flipkart.varadhi.authz.AuthorizationOptions;
 import com.flipkart.varadhi.authz.AuthorizationProvider;
-import com.flipkart.varadhi.config.DefaultAuthorizationConfiguration;
+import com.flipkart.varadhi.config.DefaultAuthorizationConfig;
 import com.flipkart.varadhi.entities.auth.*;
 import com.flipkart.varadhi.services.IamPolicyService;
 import com.flipkart.varadhi.spi.db.IamPolicyMetaStore;
@@ -25,7 +25,7 @@
 @Slf4j
 public class DefaultAuthorizationProvider implements AuthorizationProvider {
  private static final Role EMPTY_ROLE = new Role("", Set.of());
- private DefaultAuthorizationConfiguration configuration;
+ private DefaultAuthorizationConfig configuration;
  private IamPolicyService iamPolicyService;
  private volatile boolean initialised = false;
 
@@ -34,7 +34,7 @@ public Future<Boolean> init(AuthorizationOptions authorizationOptions) {
  if (!this.initialised) {
  this.configuration =
  YamlLoader.loadConfig(
- authorizationOptions.getConfigFile(), DefaultAuthorizationConfiguration.class);
+ authorizationOptions.getConfigFile(), DefaultAuthorizationConfig.class);
  getAuthZService();
  this.initialised = true;
  }

core/src/main/java/com/flipkart/varadhi/core/cluster/ClusterManager.java → server/src/main/java/com/flipkart/varadhi/cluster/ClusterManager.java

@@ -1,4 +1,4 @@
-package com.flipkart.varadhi.core.cluster;
+package com.flipkart.varadhi.cluster;
 
 import java.util.List;
 
@@ -8,9 +8,6 @@
  * be notified of the same.
  */
 public interface ClusterManager {
- void join();
-
- void leave();
 
  List<NodeInfo> getAllMembers();
 

core/src/main/java/com/flipkart/varadhi/core/cluster/MembershipListener.java → server/src/main/java/com/flipkart/varadhi/cluster/MembershipListener.java

@@ -1,4 +1,4 @@
-package com.flipkart.varadhi.core.cluster;
+package com.flipkart.varadhi.cluster;
 
 public interface MembershipListener {
  void joined(NodeInfo nodeInfo);

core/src/main/java/com/flipkart/varadhi/core/cluster/NodeConnection.java → server/src/main/java/com/flipkart/varadhi/cluster/NodeConnection.java

@@ -1,4 +1,4 @@
-package com.flipkart.varadhi.core.cluster;
+package com.flipkart.varadhi.cluster;
 
 import io.vertx.core.Future;
 import io.vertx.core.eventbus.DeliveryOptions;

core/src/main/java/com/flipkart/varadhi/core/cluster/NodeInfo.java → server/src/main/java/com/flipkart/varadhi/cluster/NodeInfo.java

@@ -1,4 +1,4 @@
-package com.flipkart.varadhi.core.cluster;
+package com.flipkart.varadhi.cluster;
 
 /**
  * TODO: need to see if the id needs to be separate from the 'ip' & 'hostname'. Implementer needs to address this.

server/src/main/java/com/flipkart/varadhi/cluster/NodeResources.java

@@ -0,0 +1,4 @@
+package com.flipkart.varadhi.cluster;
+
+public record NodeResources(int cpuCount, int nicMBps) {
+}

server/src/main/java/com/flipkart/varadhi/cluster/custom/ZookeeperClusterManager.java

@@ -0,0 +1,51 @@
+package com.flipkart.varadhi.cluster.custom;
+
+import com.flipkart.varadhi.cluster.NodeResources;
+import com.flipkart.varadhi.config.ZookeeperConnectConfig;
+import com.flipkart.varadhi.utils.CuratorFrameworkCreator;
+import io.vertx.core.Promise;
+import io.vertx.core.json.JsonObject;
+import io.vertx.core.spi.cluster.NodeInfo;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Customized zkClusterManager that only works with provided zk configuration. The curatorFramework, needs to be
+ * configured with namespace for the purpose of cluster nodes.
+ *
+ * In the future, we will customize this so that curator is not required to be namespace aware, instead this class will
+ * handle the namespacing all the paths. This should allow usage of curator client for other purposes other than just cluster
+ * management. We also need to remove curator.close() from leave method, in case the curator instance is provided.
+ *
+ * This class also customizes 1 method:
+ * 1. setNodeInfo() - to include {@link NodeResources} as part of the nodeInfo
+ */
+@Slf4j
+public class ZookeeperClusterManager extends io.vertx.spi.cluster.zookeeper.ZookeeperClusterManager {
+
+ /**
+ * Node specific resource information to be used for load-balancing. This will be persisted as part of nodeInfo
+ */
+ private final NodeResources nodeResources;
+
+ public ZookeeperClusterManager(ZookeeperConnectConfig zkConnectConfig, String nodeId, NodeResources nodeResources) {
+ super(CuratorFrameworkCreator.create(zkConnectConfig), nodeId);
+ this.nodeResources = nodeResources;
+ }
+
+ @Override
+ public void setNodeInfo(NodeInfo nodeInfo, Promise<Void> promise) {
+ NodeInfo copy = new NodeInfo(nodeInfo.host(), nodeInfo.port(),
+ nodeInfo.metadata() == null ? new JsonObject() : nodeInfo.metadata().copy()
+ );
+ copy.metadata().put("cpuCount", nodeResources.cpuCount());
+ copy.metadata().put("nicMBps", nodeResources.nicMBps());
+ super.setNodeInfo(copy, promise);
+ }
+
+ public NodeResources toNodeResources(NodeInfo nodeInfo) {
+ return new NodeResources(
+ nodeInfo.metadata().getInteger("cpuCount"),
+ nodeInfo.metadata().getInteger("nicMBps")
+ );
+ }
+}

server/src/main/java/com/flipkart/varadhi/cluster/impl/ClusterManagerImpl.java

@@ -0,0 +1,30 @@
+package com.flipkart.varadhi.cluster.impl;
+
+import com.flipkart.varadhi.cluster.ClusterManager;
+import com.flipkart.varadhi.cluster.MembershipListener;
+import com.flipkart.varadhi.cluster.NodeConnection;
+import com.flipkart.varadhi.cluster.NodeInfo;
+import lombok.RequiredArgsConstructor;
+
+import java.util.List;
+
+@RequiredArgsConstructor
+public class ClusterManagerImpl implements ClusterManager {
+
+ // TODO: add instance of zkClusterManager & clusteredVertx and use it to implement the methods of this class
+
+ @Override
+ public List<NodeInfo> getAllMembers() {
+ return null;
+ }
+
+ @Override
+ public void addMembershipListener(MembershipListener listener) {
+
+ }
+
+ @Override
+ public NodeConnection connect(String nodeId) {
+ return null;
+ }
+}

server/src/main/java/com/flipkart/varadhi/config/DefaultAuthorizationConfiguration.java → server/src/main/java/com/flipkart/varadhi/config/DefaultAuthorizationConfig.java

@@ -7,7 +7,7 @@
 import java.util.Map;
 
 @Data
-public class DefaultAuthorizationConfiguration {
+public class DefaultAuthorizationConfig {
  /**
  * role_id to Role{role_id, [permissions...]} mappings
  */

server/src/main/java/com/flipkart/varadhi/config/RestOptions.java

@@ -17,8 +17,9 @@ public class RestOptions {
  private int headersAllowedMax = HEADERS_ALLOWED_MAX;
  private int headerNameSizeMax = HEADER_NAME_SIZE_MAX;
  private int headerValueSizeMax = HEADER_VALUE_SIZE_MAX;
+
+ // TODO: These dont look related to rest. Looks related to lean deployment.
  private String defaultOrg = DEFAULT_ORG;
  private String defaultTeam = DEFAULT_TEAM;
  private String defaultProject = DEFAULT_PROJECT;
-
 }