Ray Integration to Stacks (#52)

Signed-off-by: Elamaran Shanmugam <[email protected]>

dapr-integration/README.md

@@ -7,16 +7,16 @@ Please use the following command to deploy Dapr using `idpbuilder`:
 ```bash
 idpbuilder create \
   --use-path-routing \
-  --p https://github.com/cnoe-io/stacks//dapr-integrations
+  -p https://github.com/cnoe-io/stacks//dapr-integrations \
 ```
 
 Notice that you can add Dapr to the reference implementation:
 
 ```bash
 idpbuilder create \
   --use-path-routing \
-  --p https://github.com/cnoe-io/stacks//ref-implementation
-  --p https://github.com/cnoe-io/stacks//dapr-integrations
+  -p https://github.com/cnoe-io/stacks//ref-implementation \
+  -p https://github.com/cnoe-io/stacks//dapr-integrations
 ```
 
 ## What is installed?

ray-integration/README.md

@@ -0,0 +1,31 @@
+# Ray Integrations
+
+`idpBuilder` is extensible to install Ray Operator for Serving machine learning (ML) models at scale. This integration will help us Ray Kubernetes Operator integrated into our platform which simplifies and accelerates the serving of ML models.
+
+Please use the following command to deploy ray using `idpbuilder`:
+
+```bash
+idpbuilder create \
+  --use-path-routing \
+  -p https://github.com/cnoe-io/stacks//ray-integration
+```
+
+Notice that you can add Ray to the reference implementation:
+
+```bash
+idpbuilder create \
+  --use-path-routing \
+  -p https://github.com/cnoe-io/stacks//ref-implementation \
+  -p https://github.com/cnoe-io/stacks//ray-integration
+```
+
+## What is installed?
+
+1. Ray Operator CRDs
+2. Ray Operator
+
+Once installed, you will have Ray Operator and Ray Serve components to serve the ML model and LLMs.
+
+For more information, check our module [here](https://catalog.us-east-1.prod.workshops.aws/modernengg/en-US/60-aimldelivery/63-section4-ml-model-use-case) for a step by step instructions on Serving ML Models via Internal Developer Platforms with an example. 
+
+

ray-integration/ray-operator-crds.yaml

@@ -0,0 +1,23 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Application
+metadata:
+  name: ray-operator-crds
+  namespace: argocd
+  labels:
+    env: dev
+  finalizers:
+    - resources-finalizer.argocd.argoproj.io
+spec:
+  project: default
+  source:
+    repoURL: https://github.com/ray-project/kuberay
+    targetRevision: v1.1.1
+    path: helm-chart/kuberay-operator/crds
+  destination:
+    server: "https://kubernetes.default.svc"
+  syncPolicy:
+    automated:
+      prune: true
+      selfHeal: true
+    syncOptions:
+      - Replace=true

ray-integration/ray-operator.yaml

@@ -0,0 +1,26 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Application
+metadata:
+  name: ray-operator
+  namespace: argocd
+  labels:
+    env: dev
+  finalizers:
+    - resources-finalizer.argocd.argoproj.io
+spec:
+  project: default
+  source:
+    repoURL: https://github.com/ray-project/kuberay
+    targetRevision: v1.1.1
+    path: helm-chart/kuberay-operator
+    helm:
+      skipCrds: true
+  destination:
+    server: https://kubernetes.default.svc
+    namespace: ray
+  syncPolicy:
+    automated:
+      prune: true
+      selfHeal: true
+    syncOptions:
+      - CreateNamespace=true

ref-implementation/backstage-templates/entities/catalog-info.yaml

@@ -8,6 +8,7 @@ spec:
     - ./basic/template.yaml
     - ./argo-workflows/template.yaml
     - ./app-with-bucket/template.yaml
+    - ./ray-serve/template-ray-serve.yaml
 ---
 apiVersion: backstage.io/v1alpha1
 kind: Location

ref-implementation/backstage-templates/entities/ray-serve/sample/pytorch-sample.py

@@ -0,0 +1,154 @@
+import os
+from typing import Dict
+
+import torch
+from filelock import FileLock
+from torch import nn
+from torch.utils.data import DataLoader
+from torchvision import datasets, transforms
+from torchvision.transforms import Normalize, ToTensor
+from tqdm import tqdm
+
+import ray.train
+from ray.train import ScalingConfig
+from ray.train.torch import TorchTrainer
+import ray
+
+
+def get_dataloaders(batch_size):
+    # Transform to normalize the input images
+    transform = transforms.Compose([ToTensor(), Normalize((0.5,), (0.5,))])
+
+    with FileLock(os.path.expanduser("~/data.lock")):
+        # Download training data from open datasets
+        training_data = datasets.FashionMNIST(
+            root="~/data",
+            train=True,
+            download=True,
+            transform=transform,
+        )
+
+        # Download test data from open datasets
+        test_data = datasets.FashionMNIST(
+            root="~/data",
+            train=False,
+            download=True,
+            transform=transform,
+        )
+
+    # Create data loaders
+    train_dataloader = DataLoader(training_data, batch_size=batch_size, shuffle=True)
+    test_dataloader = DataLoader(test_data, batch_size=batch_size)
+
+    return train_dataloader, test_dataloader
+
+
+# Model Definition
+class NeuralNetwork(nn.Module):
+    def __init__(self):
+        super(NeuralNetwork, self).__init__()
+        self.flatten = nn.Flatten()
+        self.linear_relu_stack = nn.Sequential(
+            nn.Linear(28 * 28, 512),
+            nn.ReLU(),
+            nn.Dropout(0.25),
+            nn.Linear(512, 512),
+            nn.ReLU(),
+            nn.Dropout(0.25),
+            nn.Linear(512, 10),
+            nn.ReLU(),
+        )
+
+    def forward(self, x):
+        x = self.flatten(x)
+        logits = self.linear_relu_stack(x)
+        return logits
+
+
+def train_func_per_worker(config: Dict):
+    lr = config["lr"]
+    epochs = config["epochs"]
+    batch_size = config["batch_size_per_worker"]
+
+    # Get dataloaders inside the worker training function
+    train_dataloader, test_dataloader = get_dataloaders(batch_size=batch_size)
+
+    # [1] Prepare Dataloader for distributed training
+    # Shard the datasets among workers and move batches to the correct device
+    # =======================================================================
+    train_dataloader = ray.train.torch.prepare_data_loader(train_dataloader)
+    test_dataloader = ray.train.torch.prepare_data_loader(test_dataloader)
+
+    model = NeuralNetwork()
+
+    # [2] Prepare and wrap your model with DistributedDataParallel
+    # Move the model to the correct GPU/CPU device
+    # ============================================================
+    model = ray.train.torch.prepare_model(model)
+
+    loss_fn = nn.CrossEntropyLoss()
+    optimizer = torch.optim.SGD(model.parameters(), lr=lr, momentum=0.9)
+
+    # Model training loop
+    for epoch in range(epochs):
+        if ray.train.get_context().get_world_size() > 1:
+            # Required for the distributed sampler to shuffle properly across epochs.
+            train_dataloader.sampler.set_epoch(epoch)
+
+        model.train()
+        for X, y in tqdm(train_dataloader, desc=f"Train Epoch {epoch}"):
+            pred = model(X)
+            loss = loss_fn(pred, y)
+
+            optimizer.zero_grad()
+            loss.backward()
+            optimizer.step()
+
+        model.eval()
+        test_loss, num_correct, num_total = 0, 0, 0
+        with torch.no_grad():
+            for X, y in tqdm(test_dataloader, desc=f"Test Epoch {epoch}"):
+                pred = model(X)
+                loss = loss_fn(pred, y)
+
+                test_loss += loss.item()
+                num_total += y.shape[0]
+                num_correct += (pred.argmax(1) == y).sum().item()
+
+        test_loss /= len(test_dataloader)
+        accuracy = num_correct / num_total
+
+        # [3] Report metrics to Ray Train
+        # ===============================
+        ray.train.report(metrics={"loss": test_loss, "accuracy": accuracy})
+
+
+def train_fashion_mnist(num_workers=5, use_gpu=False):
+    global_batch_size = 32
+
+    train_config = {
+        "lr": 1e-3,
+        "epochs": 1, # artificially set low to finish quickly
+        "batch_size_per_worker": global_batch_size // num_workers,
+    }
+
+    # Configure computation resources
+    scaling_config = ScalingConfig(num_workers=num_workers, use_gpu=use_gpu)
+
+    # Initialize a Ray TorchTrainer
+    trainer = TorchTrainer(
+        train_loop_per_worker=train_func_per_worker,
+        train_loop_config=train_config,
+        scaling_config=scaling_config,
+    )
+
+    # [4] Start distributed training
+    # Run `train_func_per_worker` on all workers
+    # =============================================
+    result = trainer.fit()
+    print(f"Training result: {result}")
+
+
+if __name__ == "__main__":
+    ray.init("auto")
+    train_fashion_mnist(num_workers=10, use_gpu=False)

ref-implementation/backstage-templates/entities/ray-serve/skeleton/catalog-info.yaml

@@ -0,0 +1,18 @@
+---
+apiVersion: backstage.io/v1alpha1
+kind: Component
+metadata:
+  name: ${{values.name | dump}}
+  description: This is for Ray Serve
+  annotations:
+    backstage.io/kubernetes-label-selector: 'entity-id=${{values.name}}'
+    backstage.io/kubernetes-namespace: argo
+    argocd/app-name: ${{values.name | dump}}
+    argo-workflows.cnoe.io/label-selector: env=dev,entity-id=${{values.name}}
+    argo-workflows.cnoe.io/cluster-name: local
+    apache-ray.cnoe.io/label-selector: env=dev,entity-id=${{values.name}}
+    apache-ray.cnoe.io/cluster-name: local
+spec:
+  owner: guest
+  lifecycle: experimental
+  type: service

ref-implementation/backstage-templates/entities/ray-serve/skeleton/manifests/ray-serve.yaml

@@ -0,0 +1,92 @@
+# Make sure to increase resource requests and limits before using this example in production.
+apiVersion: ray.io/v1
+kind: RayService
+metadata:
+  name: "ray-service-${{values.name}}"
+spec:
+  # serveConfigV2 takes a yaml multi-line scalar, which should be a Ray Serve multi-application config. See https://docs.ray.io/en/latest/serve/multi-app.html.
+  serveConfigV2: |
+    applications:
+      - name: text_ml_app
+        import_path: text_ml.app
+        route_prefix: /summarize_translate
+        runtime_env:
+          working_dir: "${{values.rayServeFile}}"
+          pip:
+            - torch
+            - transformers
+        deployments:
+          - name: Translator
+            num_replicas: 4
+            ray_actor_options:
+              num_cpus: 0.2
+            user_config:
+              language: french
+          - name: Summarizer
+            num_replicas: 4
+            ray_actor_options:
+              num_cpus: 0.2
+  rayClusterConfig:
+    rayVersion: '2.34.0' # should match the Ray version in the image of the containers
+    enableInTreeAutoscaling: true
+    autoscalerOptions:
+      upscalingMode: Conservative
+      idleTimeoutSeconds: 120
+    ######################headGroupSpecs#################################
+    # Ray head pod template.
+    headGroupSpec:
+      # The `rayStartParams` are used to configure the `ray start` command.
+      # See https://github.com/ray-project/kuberay/blob/master/docs/guidance/rayStartParams.md for the default settings of `rayStartParams` in KubeRay.
+      # See https://docs.ray.io/en/latest/cluster/cli.html#ray-start for all available options in `rayStartParams`.
+      rayStartParams:
+        dashboard-host: '0.0.0.0'
+      #pod template
+      template:
+        spec:
+          containers:
+            - name: ray-head
+              image: rayproject/ray:2.34.0
+              resources:
+                limits:
+                  cpu: 1
+                  memory: 1Gi
+                requests:
+                  cpu: 1
+                  memory: 1Gi
+              ports:
+                - containerPort: 6379
+                  name: gcs-server
+                - containerPort: 8265 # Ray dashboard
+                  name: dashboard
+                - containerPort: 10001
+                  name: client
+                - containerPort: 8000
+                  name: serve
+    workerGroupSpecs:
+      # the pod replicas in this group typed worker
+      - replicas: 1
+        minReplicas: 1
+        maxReplicas: 5
+        # logical group name, for this called small-group, also can be functional
+        groupName: small-group
+        # The `rayStartParams` are used to configure the `ray start` command.
+        # See https://github.com/ray-project/kuberay/blob/master/docs/guidance/rayStartParams.md for the default settings of `rayStartParams` in KubeRay.
+        # See https://docs.ray.io/en/latest/cluster/cli.html#ray-start for all available options in `rayStartParams`.
+        rayStartParams: {}
+        #pod template
+        template:
+          spec:
+            containers:
+              - name: ray-worker # must consist of lower case alphanumeric characters or '-', and must start and end with an alphanumeric character (e.g. 'my-name',  or '123-abc'
+                image: rayproject/ray:2.34.0
+                lifecycle:
+                  preStop:
+                    exec:
+                      command: ["/bin/sh","-c","ray stop"]
+                resources:
+                  limits:
+                    cpu: "1"
+                    memory: "4Gi"
+                  requests:
+                    cpu: "500m"
+                    memory: "2Gi"