hw7

Dockerfile

@@ -0,0 +1,12 @@
+FROM python:3.11-slim
+
+RUN pip install poetry
+
+WORKDIR /app
+
+COPY pyproject.toml poetry.lock* /app/
+RUN poetry config virtualenvs.create false && poetry install --no-dev --no-interaction --no-ansi
+
+COPY . /app
+
+CMD ["mlserver", "start", "."]

Makefile

@@ -0,0 +1,116 @@
+# Set Docker registry and image name for easier reuse
+CONTAINER_REGISTRY = dkopylov
+IMAGE_NAME = hw7
+
+# Define a Python virtual environment for dependency isolation
+venv:
+	python -m venv venv
+	source venv/bin/activate
+
+# Train the ML model inside a virtual environment
+train: venv
+	python train.py
+
+# Build Docker image with MLServer for serving the model
+build: venv
+	python shoose.py default
+	mlserver build . -t $(CONTAINER_REGISTRY)/$(IMAGE_NAME)
+
+build-alt: venv
+	python choose.py alt
+	mlserver build . -t $(CONTAINER_REGISTRY)/$(IMAGE_NAME)
+
+# Push Docker image to registry
+push-image:
+	docker push $(CONTAINER_REGISTRY)/$(IMAGE_NAME)
+
+# Create a Kind cluster for local Kubernetes testing
+kind-cluster:
+	kind create cluster --name seldon-cluster --config kind-cluster.yaml --image=kindest/node:v1.21.2
+
+# Install Ambassador as an API gateway using Helm
+ambassador: kind-cluster
+	helm repo add datawire https://www.getambassador.io
+	helm repo update
+	helm upgrade --install ambassador datawire/ambassador \
+		--set image.repository=docker.io/datawire/ambassador \
+		--set service.type=ClusterIP \
+		--set replicaCount=1 \
+		--set crds.keep=false \
+		--set enableAES=false \
+		--create-namespace \
+		--namespace ambassador
+
+# Install Seldon Core in Kubernetes to manage ML deployments
+seldon-core: kind-cluster
+	helm repo add seldon https://storage.googleapis.com/seldon-charts
+	helm repo update
+	helm upgrade --install seldon-core seldon/seldon-core-operator \
+		--set crd.create=true \
+		--set usageMetrics.enabled=true \
+		--set ambassador.enabled=true \
+		--create-namespace \
+		--namespace seldon-system
+
+# Install Prometheus for monitoring via Helm
+prometheus: kind-cluster
+	helm repo add prometheus-community https://prometheus-community.github.io/helm-charts
+	helm repo update
+	helm upgrade --install seldon-monitoring prometheus-community/kube-prometheus-stack \
+		--set fullnameOverride=seldon-monitoring \
+		--create-namespace \
+		--namespace seldon-monitoring
+
+# Install Grafana for metrics visualization via Helm
+grafana: kind-cluster
+	helm repo add grafana https://grafana.github.io/helm-charts
+	helm repo update
+	helm upgrade --install grafana-seldon-monitoring grafana/grafana \
+		--set version=6.56.1 \
+		--values service.yaml \
+		--namespace seldon-monitoring
+
+# Deploy all components
+deploy-everything: kind-cluster ambassador seldon-core prometheus grafana
+
+# Deploy the application using Seldon to manage the machine learning lifecycle
+deploy:
+	kubectl apply -f seldondeployment.yaml
+	kubectl apply -f podmonitor.yaml
+
+# Delete the deployed predictor application
+delete-deployment:
+	kubectl delete -f seldondeployment.yaml
+
+# Port forwarding commands for local access
+port-forward-grafana:
+	kubectl port-forward svc/grafana-seldon-monitoring 3000:80 --namespace seldon-monitoring
+
+port-forward-prometheus:
+	kubectl port-forward svc/seldon-monitoring-prometheus 9090 --namespace seldon-monitoring
+
+# Test the deployment by sending a test request
+test-request:
+	curl -X POST -H "Content-Type: application/json" \
+        -d '{"data": {"ndarray": [[your_features_here]]}}' \
+        http://localhost:8000/seldon/default/$(MODEL_NAME)/api/v1.0/predictions
+
+# Display help for available Makefile commands
+help:
+	@echo "Available commands:"
+	@echo "train                   - Train the model."
+	@echo "build                   - Build Docker image with MLServer (one)."
+    @echo "build-alt               - Build Docker image with MLServer (two)."
+	@echo "push-image              - Push Docker image to registry."
+	@echo "kind-cluster            - Create a Kind cluster."
+	@echo "ambassador              - Install Ambassador."
+	@echo "seldon-core             - Install Seldon Core."
+	@echo "prometheus              - Install Prometheus."
+	@echo "grafana                 - Install Grafana."
+	@echo "deploy-everything       - Deploy all infrastructure components."
+	@echo "deploy                  - Deploy the application."
+	@echo "delete-deployment       - Delete the application deployment."
+	@echo "port-forward-grafana    - Port-forward for Grafana."
+	@echo "port-forward-prometheus - Port-forward for Prometheus."
+	@echo "test-request            - Send a test request."
+	@echo "help                    - Display this help message."

choose.py

@@ -0,0 +1,58 @@
+"""Choose model to build"""
+import json
+import sys
+import os
+from typing import Dict, List
+
+def update_model_settings(model_type: str) -> None:
+    """
+    Updates the model settings file based on the specified model type.
+
+    Args:
+        model_type (str): The type of model configuration to apply. Valid options are 'solo' or 'two'.
+
+    Raises:
+        ValueError: If an invalid model type is specified.
+        FileNotFoundError: If the settings file does not exist and cannot be removed.
+    """
+    settings_path = "model-config.json"
+
+    # Define basic structure of model configuration data
+    configuration: Dict[str, any] = {
+        "name": "uplift-predictor",
+        "implementation": "inference.UpliftModel",
+        "parameters": {"uri": ""}
+    }
+
+    # Assign the correct model URI based on the command line argument
+    if model_type == "default":
+        configuration["parameters"]["uri"] = "one_model.joblib"
+    elif model_type == "alt":
+        configuration["parameters"]["uri"] = "two_model.joblib"
+    else:
+        raise ValueError("Invalid model type specified. Choose either 'default' or 'alt'.")
+
+    # Attempt to remove the existing settings file if it exists
+    try:
+        os.remove(settings_path)
+    except FileNotFoundError:
+        print(f"Notice: The file {settings_path} was not found and will be created.")
+
+    # Write the updated configuration to the file
+    with open(settings_path, "w") as file:
+        json.dump(configuration, file, indent=4)
+    print(f"Model settings updated successfully in {settings_path}.")
+
+def main() -> None:
+    """
+    Main function that processes command line arguments to update model settings.
+    """
+    if len(sys.argv) != 2 or sys.argv[1] not in ["default", "alt"]:
+        print("Usage error: The script takes exactly 1 argument: {default, alt}")
+        sys.exit(-1)
+
+    # Update model settings based on the provided argument
+    update_model_settings(sys.argv[1])
+
+if __name__ == "__main__":
+    main()

inference.py

@@ -0,0 +1,78 @@
+from mlserver import MLModel, types
+from mlserver.utils import get_model_uri
+from mlserver.codecs import StringCodec
+from joblib import load
+from typing import Dict, Any
+import numpy as np
+import json
+
+
+class UpliftModel(MLModel):
+    """
+    An asynchronous ML server model for predicting customer uplift using an uplift model.
+    """
+    async def initialize_model(self) -> bool:
+        """
+        Asynchronously loads the model from a URI specified in the settings.
+
+        Returns:
+            bool: True if the model is successfully loaded and ready, False otherwise.
+        """
+        model_path = await get_model_uri(self._settings)
+        self.uplift_estimator = load(model_path)
+        self.is_ready = True
+        return self.is_ready
+
+    async def perform_prediction(self, request: types.InferenceRequest) -> types.InferenceResponse:
+        """
+        Process an inference request and returns the prediction results.
+
+        Args:
+            request (types.InferenceRequest): The inference request object containing input data.
+
+        Returns:
+            types.InferenceResponse: The response object containing the prediction results.
+        """
+        try:
+            decoded_request = self._parse_request(request).get("predict_request", {})
+            feature_array = np.array(decoded_request.get("data", []))
+
+            prediction_result = {"success": True, "prediction": self.uplift_estimator.predict(feature_array)}
+
+        except Exception as e:
+            prediction_result = {"success": False, "prediction": None}
+
+        response_payload = json.dumps(prediction_result.__repr__()).encode("UTF-8")
+
+        return types.InferenceResponse(
+            id=request.id,
+            model_name=self.name,
+            model_version=self.version,
+            outputs=[
+                types.ResponseOutput(
+                    name="prediction_output",
+                    shape=[len(response_payload)],
+                    datatype="BYTES",
+                    data=[response_payload],
+                    parameters=types.Parameters(content_type="application/json"),
+                )
+            ],
+        )
+
+    def _parse_request(self, request: types.InferenceRequest) -> Dict[str, Any]:
+        """
+        Decodes and extracts JSON data from an inference request's inputs.
+
+        Args:
+            request (types.InferenceRequest): The request from which to extract data.
+
+        Returns:
+            Dict[str, Any]: A dictionary containing the decoded data.
+        """
+        decoded_inputs = {}
+        for input_data in request.inputs:
+            decoded_inputs[input_data.name] = json.loads(
+                "".join(self.decode(input_data, default_codec=StringCodec))
+            )
+
+        return decoded_inputs

kind-cluster.yaml

@@ -0,0 +1,34 @@
+kind: Cluster
+apiVersion: kind.x-k8s.io/v1alpha4
+nodes:
+  - role: control-plane
+  - role: worker
+    kubeadmConfigPatches:
+      - |
+        kind: JoinConfiguration
+        nodeRegistration:
+          kubeletExtraArgs:
+            node-labels: "node=worker_1"
+        extraMounts:
+          - hostPath: ./data
+        containerPath: /tmp/data
+  - role: worker
+    kubeadmConfigPatches:
+      - |
+        kind: JoinConfiguration
+        nodeRegistration:
+          kubeletExtraArgs:
+            node-labels: "node=worker_2"
+        extraMounts:
+          - hostPath: ./data
+        containerPath: /tmp/data
+  - role: worker
+    kubeadmConfigPatches:
+      - |
+        kind: JoinConfiguration
+        nodeRegistration:
+          kubeletExtraArgs:
+            node-labels: "node=worker_3"
+        extraMounts:
+          - hostPath: ./data
+        containerPath: /tmp/data

model-settings.json

@@ -0,0 +1,8 @@
+{
+    "name": "uplift-predictor",
+    "implementation": "UpliftPredictor",
+    "parameters": {
+      "uri": "./uplift_model.joblib"
+    }
+  }
+

podmonitor.yaml

@@ -0,0 +1,28 @@
+apiVersion: monitoring.coreos.com/v1
+kind: PodMonitor
+metadata:
+  name: custom-uplift-monitor
+  namespace: uplift-metrics
+  labels:
+    monitor: uplift-pod
+  annotations:
+    description: "PodMonitor for monitoring the uplift prediction service managed by Seldon"
+spec:
+  selector:
+    matchLabels:
+      app.kubernetes.io/name: uplift-service
+      app.kubernetes.io/part-of: uplift-pipeline
+  podMetricsEndpoints:
+    - port: metrics-port
+      path: /metrics
+      interval: 30s
+      scheme: http
+      honorLabels: true
+      scrapeTimeout: 10s
+      metricRelabelings:
+        - sourceLabels: [__name__]
+          regex: 'process.*'
+          action: keep
+  namespaceSelector:
+    matchNames:
+      - uplift-metrics