feat(sherlock): Implemented the inferencing API

isala404 · Mar 10, 2022 · b2f07b5 · b2f07b5
1 parent ae09956
commit b2f07b5
Show file tree

Hide file tree

Showing 4 changed files with 146 additions and 0 deletions.
diff --git a/.github/workflows/build-sherlock.yaml b/.github/workflows/build-sherlock.yaml
@@ -0,0 +1,28 @@
+name: sherlock
+
+on:
+  push:
+    branches:
+      - main
+
+env:
+  IMAGE_NAME: ghcr.io/mrsupiri/lazy-koala/sherlock
+  DOCKER_BUILDKIT: 1
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        shell: bash
+        working-directory: sherlock
+    steps:
+      - uses: actions/checkout@v2
+      - uses: docker/login-action@v1
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+      - run: docker build -t $IMAGE_NAME:latest .
+      - run: docker tag $IMAGE_NAME:latest $IMAGE_NAME:commit-${GITHUB_SHA:0:8}
+      - run: docker push $IMAGE_NAME --all-tags
diff --git a/sherlock/Cargo.toml b/sherlock/Cargo.toml
@@ -0,0 +1,13 @@
+[package]
+name = "sherlock"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+prometheus = "0.13.0"
+lazy_static = "1.4.0"
+hyper = { version = "0.14", features = ["full"] }
+tokio = { version = "1", features = ["full"] }
+reqwest = { version = "0.11", features = ["blocking", "json"] }
diff --git a/sherlock/Dockerfile b/sherlock/Dockerfile
@@ -0,0 +1,28 @@
+# Build the binary
+FROM ekidd/rust-musl-builder as builder
+WORKDIR /sherlock
+USER root
+
+# Copy the source and build the application.
+COPY Cargo.toml ./
+COPY src ./src
+RUN cargo build --target x86_64-unknown-linux-musl --release
+
+
+# Production Image
+FROM alpine:latest
+RUN apk --no-cache add ca-certificates
+
+# COPY the binary
+COPY --from=builder /sherlock/target/x86_64-unknown-linux-musl/release/sherlock /sherlock
+
+# DEFAULT ENV
+ENV SERVICE_NAME=sample_service
+ENV NAMESPACE=default
+ENV END_POINT=http://localhost:8501/v1/models/sherlock:predict
+ENV POOL_DURATION=60
+
+# Define port
+EXPOSE 9898
+
+CMD ["/sherlock"]
diff --git a/sherlock/src/main.rs b/sherlock/src/main.rs
@@ -0,0 +1,77 @@
+use hyper::{
+    header::CONTENT_TYPE,
+    service::{make_service_fn, service_fn},
+    Body, Request, Response, Server,
+};
+use prometheus::{Encoder, Gauge, TextEncoder};
+use std::{env::var, thread, time::Duration};
+use lazy_static::lazy_static;
+use prometheus::{labels, opts, register_gauge};
+use std::collections::HashMap;
+
+lazy_static! {
+    static ref SERVICE_NAME: String = var("SERVICE_NAME").unwrap();
+    static ref NAMESPACE: String = var("NAMESPACE").unwrap();
+    static ref END_POINT: String = var("END_POINT").unwrap();
+    static ref POOL_DURATION: String = var("POOL_DURATION").unwrap();
+    static ref ANOMLAY_GAUGE: Gauge = register_gauge!(opts!(
+        "anomaly_score",
+        "Reconstruction loss of the autoencoder",
+        labels! {"serviceName" => SERVICE_NAME.as_str(), "namespace" => NAMESPACE.as_str()}
+    ))
+    .unwrap();
+}
+
+async fn serve_req(_req: Request<Body>) -> Result<Response<Body>, hyper::Error> {
+    let encoder = TextEncoder::new();
+
+    let metric_families = prometheus::gather();
+    let mut buffer = vec![];
+    encoder.encode(&metric_families, &mut buffer).unwrap();
+
+    let response = Response::builder()
+        .status(200)
+        .header(CONTENT_TYPE, encoder.format_type())
+        .body(Body::from(buffer))
+        .unwrap();
+
+    Ok(response)
+}
+
+
+fn query_model() ->Result<f64, Box<dyn std::error::Error>> {
+    let resp = reqwest::blocking::get(END_POINT.as_str())?
+                .json::<HashMap<String, f64>>()?;
+    Ok(resp["predictions"])
+}
+
+fn poll_anomaly_scores(delay: u64) {
+
+    loop {
+
+        if let Ok(value) = query_model(){
+            ANOMLAY_GAUGE.set(value);
+        }
+
+        thread::sleep(Duration::from_secs(delay));
+    }
+}
+
+#[tokio::main]
+async fn main() {
+
+    thread::spawn(|| poll_anomaly_scores(POOL_DURATION.as_str().parse::<u64>().unwrap()));
+
+    ANOMLAY_GAUGE.set(0.0);
+
+    let addr = ([0, 0, 0, 0], 9898).into();
+    println!("Listening on http://{}", addr);
+
+    let serve_future = Server::bind(&addr).serve(make_service_fn(|_| async {
+        Ok::<_, hyper::Error>(service_fn(serve_req))
+    }));
+
+    if let Err(err) = serve_future.await {
+        eprintln!("server error: {}", err);
+    }
+}