kserve/docs/samples/drift-detection/alibi-detect/cifar10/cifar10_drift.ipynb

Cifar10 Drift Detection

In this example we will deploy an image classification model along with a drift detector trained on the same dataset. For in depth details on creating a drift detection model for your own dataset see the alibi-detect project and associated documentation. You can find details for this CIFAR10 example in their documentation as well.

Prequisites:

• Running cluster with
  • kfserving installed
  • Knative eventing installed >= 0.18

Tested on GKE and Kind with Knative Eventing 0.18 and Istio 1.7.3

In [ ]:

!pip install -r requirements_notebook.txt

Setup Resources

In [8]:

!kubectl create namespace cifar10

namespace/cifar10 created

In [9]:

%%writefile broker.yaml
apiVersion: eventing.knative.dev/v1
kind: broker
metadata:
 name: default
 namespace: cifar10

Overwriting broker.yaml

In [10]:

!kubectl create -f broker.yaml

broker.eventing.knative.dev/default created

In [11]:

%%writefile event-display.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: hello-display
  namespace: cifar10
spec:
  replicas: 1
  selector:
    matchLabels: &labels
      app: hello-display
  template:
    metadata:
      labels: *labels
    spec:
      containers:
        - name: event-display
          image: gcr.io/knative-releases/knative.dev/eventing-contrib/cmd/event_display

---

kind: Service
apiVersion: v1
metadata:
  name: hello-display
  namespace: cifar10
spec:
  selector:
    app: hello-display
  ports:
  - protocol: TCP
    port: 80
    targetPort: 8080

Overwriting event-display.yaml

In [12]:

!kubectl apply -f event-display.yaml

deployment.apps/hello-display created
service/hello-display created

Create the Kfserving image classification model for Cifar10. We add in a logger for requests - the default destination is the namespace Knative Broker.

In [13]:

%%writefile cifar10.yaml
apiVersion: "serving.kserve.io/v1beta1"
kind: "InferenceService"
metadata:
  name: "tfserving-cifar10"
  namespace: cifar10
spec:
  predictor:
    tensorflow:
      storageUri: "gs://seldon-models/tfserving/cifar10/resnet32"
    logger:
      mode: all
      url: http://broker-ingress.knative-eventing.svc.cluster.local/cifar10/default

Overwriting cifar10.yaml

In [14]:

!kubectl apply -f cifar10.yaml

inferenceservice.serving.kserve.io/tfserving-cifar10 created

Create the pretrained Drift Detector. We forward replies to the message-dumper we started. Notice the drift_batch_size. The drift detector will wait until drift_batch_size number of requests are received before making a drift prediction.

In [15]:

%%writefile cifar10cd.yaml
apiVersion: serving.knative.dev/v1
kind: Service
metadata:
  name: drift-detector
  namespace: cifar10
spec:
  template:
    metadata:
      annotations:
        autoscaling.knative.dev/minScale: "1"
    spec:
      containers:
      - image: seldonio/alibi-detect-server:0.0.2
        imagePullPolicy: IfNotPresent
        args:
        - --model_name
        - cifar10cd
        - --http_port
        - '8080'
        - --protocol
        - tensorflow.http
        - --storage_uri
        - gs://seldon-models/alibi-detect/cd/ks/cifar10
        - --reply_url
        - http://hello-display.cifar10
        - --event_type
        - org.kubeflow.serving.inference.outlier
        - --event_source
        - org.kubeflow.serving.cifar10cd
        - DriftDetector
        - --drift_batch_size
        - '5000'

Overwriting cifar10cd.yaml

In [29]:

!kubectl apply -f cifar10cd.yaml

service.serving.knative.dev/drift-detector configured

Create a Knative trigger to forward logging events to our Outlier Detector.

In [30]:

%%writefile trigger.yaml
apiVersion: eventing.knative.dev/v1
kind: Trigger
metadata:
  name: drift-trigger
  namespace: cifar10
spec:
  broker: default
  filter:
    attributes:
      type: org.kubeflow.serving.inference.request
  subscriber:
    ref:
      apiVersion: serving.knative.dev/v1
      kind: Service
      name: drift-detector
      namespace: cifar10

Overwriting trigger.yaml

In [31]:

!kubectl apply -f trigger.yaml

trigger.eventing.knative.dev/drift-trigger created

Get the IP address of the Istio Ingress Gateway. This assumes you have installed istio with a LoadBalancer.

In [ ]:

CLUSTER_IPS = !(kubectl -n istio-system get service istio-ingressgateway -o jsonpath='{.status.loadBalancer.ingress[0].ip}')
CLUSTER_IP = CLUSTER_IPS[0]
print(CLUSTER_IP)

If you are using Kind or Minikube you will need to port-forward to the istio ingressgateway and uncomment the following cell.

INGRESS_GATEWAY_SERVICE=$(kubectl get svc --namespace istio-system --selector="app=istio-ingressgateway" --output jsonpath='{.items[0].metadata.name}')
kubectl port-forward --namespace istio-system svc/${INGRESS_GATEWAY_SERVICE} 8080:80

In [24]:

# CLUSTER_IP="localhost:8080"

In [20]:

SERVICE_HOSTNAMES = !(kubectl get inferenceservice -n cifar10 tfserving-cifar10 -o jsonpath='{.status.url}' | cut -d "/" -f 3)
SERVICE_HOSTNAME_CIFAR10 = SERVICE_HOSTNAMES[0]
print(SERVICE_HOSTNAME_CIFAR10)

tfserving-cifar10.cifar10.example.com

In [21]:

SERVICE_HOSTNAMES = !(kubectl get ksvc -n cifar10 drift-detector -o jsonpath='{.status.url}' | cut -d "/" -f 3)
SERVICE_HOSTNAME_VAEOD = SERVICE_HOSTNAMES[0]
print(SERVICE_HOSTNAME_VAEOD)

drift-detector.cifar10.example.com

In [22]:

import matplotlib.pyplot as plt
import numpy as np
import requests
import json
import tensorflow as tf

tf.keras.backend.clear_session()

train, test = tf.keras.datasets.cifar10.load_data()
X_train, y_train = train
X_test, y_test = test

X_train = X_train.astype("float32") / 255
X_test = X_test.astype("float32") / 255
print(X_train.shape, y_train.shape, X_test.shape, y_test.shape)
classes = (
    "plane",
    "car",
    "bird",
    "cat",
    "deer",
    "dog",
    "frog",
    "horse",
    "ship",
    "truck",
)


def show(X):
    plt.imshow(X.reshape(32, 32, 3))
    plt.axis("off")
    plt.show()


def predict(X):
    formData = {"instances": X.tolist()}
    headers = {}
    headers["Host"] = SERVICE_HOSTNAME_CIFAR10
    res = requests.post(
        "http://" + CLUSTER_IP + "/v1/models/tfserving-cifar10:predict",
        json=formData,
        headers=headers,
    )
    if res.status_code == 200:
        j = res.json()
        if len(j["predictions"]) == 1:
            return classes[np.array(j["predictions"])[0].argmax()]
    else:
        print("Failed with ", res.status_code)
        return []


def drift(X):
    formData = {"instances": X.tolist()}
    headers = {
        "ce-namespace": "default",
        "ce-modelid": "cifar10drift",
        "ce-type": "io.seldon.serving.inference.request",
        "ce-id": "1234",
        "ce-source": "localhost",
        "ce-specversion": "1.0",
    }
    headers["Host"] = SERVICE_HOSTNAME_VAEOD
    res = requests.post("http://" + CLUSTER_IP + "/", json=formData, headers=headers)
    if res.status_code == 200:
        od = res.json()
        return od
    else:
        print("Failed with ", res.status_code)
        return []

(50000, 32, 32, 3) (50000, 1) (10000, 32, 32, 3) (10000, 1)

Normal Prediction

In [32]:

idx = 1
X = X_train[idx : idx + 1]
show(X)
predict(X)

Out [32]:

'truck'

Test Drift

We need to accumulate a large enough batch size so no drift will be tested as yet.

In [25]:

!kubectl logs -n cifar10 $(kubectl get pod -n cifar10 -l app=hello-display -o jsonpath='{.items[0].metadata.name}')

We will now send 5000 requests to the model in batches. The drift detector will run at the end of this as we set the drift_batch_size to 5000 in our yaml above.

In [33]:

from tqdm.notebook import tqdm

for i in tqdm(range(0, 5000, 100)):
    X = X_train[i : i + 100]
    predict(X)

HBox(children=(HTML(value=''), FloatProgress(value=0.0, max=50.0), HTML(value='')))

Let's check the message dumper and extract the first drift result.

In [34]:

res = !kubectl logs -n cifar10 $(kubectl get pod -n cifar10 -l app=hello-display -o jsonpath='{.items[0].metadata.name}')
data = []
for i in range(0, len(res)):
    if res[i] == "Data,":
        data.append(res[i + 1])
j = json.loads(json.loads(data[0]))
print("Drift", j["data"]["is_drift"] == 1)

Drift False

Now, let's create some CIFAR10 examples with motion blur.

In [35]:

from alibi_detect.datasets import fetch_cifar10c, corruption_types_cifar10c

corruption = ["motion_blur"]
X_corr, y_corr = fetch_cifar10c(corruption=corruption, severity=5, return_X_y=True)
X_corr = X_corr.astype("float32") / 255

In [36]:

show(X_corr[0])
show(X_corr[1])
show(X_corr[2])

Send these examples to the predictor.

In [37]:

for i in tqdm(range(0, 5000, 100)):
    X = X_corr[i : i + 100]
    predict(X)

HBox(children=(HTML(value=''), FloatProgress(value=0.0, max=50.0), HTML(value='')))

Now when we check the message dump we should find a new drift response.

In [38]:

res = !kubectl logs -n cifar10 $(kubectl get pod -n cifar10 -l app=hello-display -o jsonpath='{.items[0].metadata.name}')
data = []
for i in range(0, len(res)):
    if res[i] == "Data,":
        data.append(res[i + 1])
j = json.loads(json.loads(data[-1]))
print("Drift", j["data"]["is_drift"] == 1)

Drift True

Tear Down

In [39]:

!kubectl delete ns cifar10

namespace "cifar10" deleted

In [ ]: