test: getting ready for testing with operator

Signed-off-by: vsoch <[email protected]>

Author: vsoch

README.md

@@ -192,6 +192,9 @@ ensemble-server start
 # Run the hello-world example ensemble! it will submit and monitor job events, etc
 ensemble run examples/hello-world.yaml
 
+# Here is how to add on the fly debug (logging->debug true)
+ensemble run --debug examples/hello-world.yaml
+
 # This example shows using repetitions and backoff
 ensemble run examples/backoff-example.yaml
 
@@ -206,11 +209,8 @@ ensemble run examples/terminate-example.yaml
 ensemble run examples/heartbeat-example.yaml
 
 # Grow/shrink requires a minicluster (flux doesn't support it) but we can mock it here
-# TODO - add greater than parsing, then run example here, then in operator
-ensemble run --executor minicluster examples/grow-shrink-example.yaml
-
-# Here is how to add on the fly debug (logging->debug true)
-ensemble run --debug --executor minicluster examples/grow-shrink-example.yaml
+# Note that a --name with <namespace>/<name> is required
+ensemble run --name default/ensemble --executor minicluster examples/grow-shrink-example.yaml
 ```
 
 Right now, this will run any rules with "start" triggers, which for this hello world example includes a few hello world jobs! You'll then be able to watch and see flux events coming in!

ensemble/client/__init__.py

@@ -46,12 +46,27 @@ def get_parser():
         choices=defaults.supported_members,
         default="flux",
     )
+    run.add_argument(
+        "--name",
+        help="Identifier for member (required for minicluster)",
+    )
     run.add_argument(
         "--debug",
         help="Enable debug logging for the config",
         action="store_true",
         default=False,
     )
+    run.add_argument(
+        "--port",
+        help=f"Port to run application (defaults to {defaults.port})",
+        default=defaults.port,
+        type=int,
+    )
+    run.add_argument(
+        "--host",
+        help="Host with server (defaults to localhost)",
+        default="localhost",
+    )
 
     for command in [run]:
         command.add_argument(

ensemble/client/run.py

@@ -2,7 +2,11 @@
 
 
 def main(args, parser, extra, subparser):
+    # Assemble options
+    options = {"name": args.name, "port": args.port, "host": args.host}
+
     # This will raise an error if the member type (e.g., minicluster) is not known
-    member = members.get_member(args.executor)
+    member = members.get_member(args.executor, options=options)
+
     member.load(args.config, args.debug)
     member.start()

ensemble/members/base.py

@@ -16,9 +16,8 @@ def __init__(self, **options):
         """
         Create a new member type (e.g., FluxQueue)
         """
-        # Set options as attributes
-        for key, value in options.items():
-            setattr(self, key, value)
+        # Set options
+        self.options = options
 
         # Common queue metrics
         self.metrics = QueueMetrics()

ensemble/members/flux/minicluster.py

@@ -13,35 +13,76 @@ class FluxMiniClusterQueue(MemberBase):
     member supports scale up and scale down for actions.
     """
 
-    # TODO for ensemble operator - we need the ensemble name to identify it
-    # perhaps generate randomly or similar?
+    def __init__(self, **kwargs):
+        """
+        Create a new flux MiniCluster
+
+        Since we need to communicate back to the grpc service, we require a name
+        and namespace. This should be provided via the name kwargs, which is parsed
+        into both. E.g,, --name default/ensemble. If a namespace is not provided,
+        default is assumed.
+        """
+        super().__init__(**kwargs)
+        self.set_identifier()
+
+    def set_identifier(self):
+        """
+        Get the name/namespace of the MiniCluster
+        """
+        if "name" not in self.options or not self.options["name"]:
+            raise ValueError("A --name (namespace/name) is required for a minicluster")
+        name = self.options["name"]
+        namespace = "default"
+        if "/" in name:
+            namespace, name = name.split("/")
+        self.options["name"] = name
+        self.options["namespace"] = namespace
+
+    @property
+    def host(self):
+        """
+        Host can be customized with options, and defaults to localhost:50051
+        """
+        host = self.options.get("host") or "localhost"
+        port = self.options.get("port") or 50051
+        return f"{host}:{port}"
 
     @property
     def client(self):
         """
         Ensure we have a connection to the service client.
         """
-        # Do we have the ensemble client?
-        # TODO allow to customize host, etc.
         if hasattr(self, "_client"):
             return self._client
-        self._client = EnsembleClient()
+        self._client = EnsembleClient(host=self.host)
         return self._client
 
     def grow(self, rule, record=None):
         """
         Request to the API to grow the MiniCluster
         """
+        # We need to send over the name and namespace
+        name = self.options["name"]
+        namespace = self.options["namespace"]
+
         # For now use the ensemble type as the name
         # TODO this needs to be caught and decided upon - retry?
-        response = self.client.action_request(self.name, "grow", {})
+        response = self.client.action_request(
+            name=name, namespace=namespace, action="grow", payload={}
+        )
         print(response)
 
     def shrink(self, rule, record=None):
         """
         Request to the API to shrink the MiniCluster
         """
-        response = self.client.action_request(self.name, "shrink", {})
+        # We need to send over the name and namespace
+        name = self.options["name"]
+        namespace = self.options["namespace"]
+
+        response = self.client.action_request(
+            name=name, namespace=namespace, action="shrink", payload={}
+        )
         print(response)
 
     @property

ensemble/members/flux/queue.py

@@ -59,7 +59,6 @@ class FluxQueue(MemberBase):
 
     def __init__(
         self,
-        refresh_interval=0,
         summary_frequency=10,
         **kwargs,
     ):
@@ -68,11 +67,9 @@ def __init__(
 
         Parameters:
         summary_frequency (int): how often (events) to show summary
-        refresh_interval (int) : number of seconds to refresh all metrics (0 indicates no refresh)
         to 60 seconds. If you set to 0, it will not be set.
         """
         self.handle = flux.Flux()
-        self.refresh_interval = refresh_interval
 
         # How often on job completions to summarize?
         self.summary_freqency = summary_frequency

ensemble/protos/ensemble_service_pb2.py

@@ -6,20 +6,24 @@ from typing import ClassVar as _ClassVar, Optional as _Optional, Union as _Union
 DESCRIPTOR: _descriptor.FileDescriptor
 
 class StatusRequest(_message.Message):
-    __slots__ = ("member",)
+    __slots__ = ("member", "namespace")
     MEMBER_FIELD_NUMBER: _ClassVar[int]
+    NAMESPACE_FIELD_NUMBER: _ClassVar[int]
     member: str
-    def __init__(self, member: _Optional[str] = ...) -> None: ...
+    namespace: str
+    def __init__(self, member: _Optional[str] = ..., namespace: _Optional[str] = ...) -> None: ...
 
 class ActionRequest(_message.Message):
-    __slots__ = ("member", "action", "payload")
+    __slots__ = ("member", "namespace", "action", "payload")
     MEMBER_FIELD_NUMBER: _ClassVar[int]
+    NAMESPACE_FIELD_NUMBER: _ClassVar[int]
     ACTION_FIELD_NUMBER: _ClassVar[int]
     PAYLOAD_FIELD_NUMBER: _ClassVar[int]
     member: str
+    namespace: str
     action: str
     payload: str
-    def __init__(self, member: _Optional[str] = ..., action: _Optional[str] = ..., payload: _Optional[str] = ...) -> None: ...
+    def __init__(self, member: _Optional[str] = ..., namespace: _Optional[str] = ..., action: _Optional[str] = ..., payload: _Optional[str] = ...) -> None: ...
 
 class Response(_message.Message):
     __slots__ = ("payload", "status")

ensemble/protos/ensemble_service_pb2.pyi

@@ -8,15 +8,15 @@
     "type": "object",
     # The only required thing is the algorithm, since we can respond to jobs that
     # are submit otherwise in the queue.
-    "required": ["rules"],
+    "required": ["rules", "jobs"],
     "properties": {
-        "attributes": {"$ref": "#/definitions/algorithm"},
         "jobs": {"$ref": "#/definitions/jobs"},
         "rules": {"$ref": "#/definitions/rules"},
+        "logging": {"$ref": "#/definitions/logging"},
+        "custom": {"type": "string"},
         "additionalProperties": False,
     },
     "definitions": {
-        "custom": {"type": "string"},
         "logging": {
             "type": "object",
             "properties": {

ensemble/schema.py

@@ -83,6 +83,7 @@ def RequestAction(self, request, context):
         Request an action is performed according to an algorithm.
         """
         print(f"Member {request.member}")
+        print(f"Namespace {request.namespace}")
         print(f"Action {request.action}")
         print(f"Payload {request.payload}")
 

ensemble/server.py

@@ -14,13 +14,15 @@ message StatusRequest {
 
     // This is the ensemble member type (e.g., minicluster)
     string member = 1;
+    string namespace = 2;
 }
 
 // ActionRequest requests an action
 message ActionRequest {
     string member = 1;
-    string action = 2;
-    string payload = 3;
+    string namespace = 2;
+    string action = 3;
+    string payload = 4;
 }
 
 message Response {