Add interface to support pause/resume gart service

Signed-off-by: Lei Wang <[email protected]>

charts/gart/templates/_helpers.tpl

@@ -38,6 +38,10 @@ If release name contains chart name it will be used as a full name.
 {{- define "gart.analyzer.fullname" -}}
 {{- printf "%s-%s" (include "gart.fullname" .) "analyzer" | trunc 63 | trimSuffix "-" -}}
 {{- end -}}
+
+{{- define "gart.controller-service.fullname" -}}
+{{- printf "%s-%s" (include "gart.fullname" .) "controller-service" | trunc 63 | trimSuffix "-" -}}
+{{- end -}}
 
 
 {{/*
@@ -135,6 +139,23 @@ Return the proper gart debezium image name
 {{- end -}}
 {{- end -}}
 
+{{/*
+Return the proper gart controller image name
+*/}}
+{{- define "gart.controller.image" -}}
+{{- $tag := .Chart.AppVersion | toString -}}
+{{- with .Values.controller.image -}}
+{{- if .tag -}}
+{{- $tag = .tag | toString -}}
+{{- end -}}
+{{- if .registry -}}
+{{- printf "%s/%s:%s" .registry .repository $tag -}}
+{{- else -}}
+{{- printf "%s:%s" .repository $tag -}}
+{{- end -}}
+{{- end -}}
+{{- end -}}
+
 {{/*
 Return the proper gart curl image name
 */}}

charts/gart/templates/configmap.yaml

@@ -1,3 +1,7 @@
+{{- $debezium_service_name := include "gart.debezium.fullname" . }}
+{{- $debezium_service_port := int .Values.debezium.containerPort }}
+{{- $debezium_service := printf "%s:%d" $debezium_service_name $debezium_service_port }}
+
 apiVersion: v1
 kind: ConfigMap
 metadata:
@@ -15,3 +19,6 @@ data:
   ETCD_PREFIX: {{ .Values.dataconfig.etcdPrefix | quote }}
   ENABLE_BULKLOAD: {{ .Values.dataconfig.enableBulkload | quote }}
   SUBGRAPH_NUM: {{ .Values.dataconfig.subgraphNum | quote }}
+  DEBEZIUM_SERVICE: {{ $debezium_service | quote }}
+  DEBEZIUM_CONNECTOR_NAME: {{ include "gart.debezium.fullname" . | quote }}
+  CONTROLLER_FLASK_PORT: {{ .Values.controller.containerPort | quote }}

charts/gart/templates/converter/deployment.yaml

@@ -76,6 +76,15 @@ spec:
         volumeMounts:
         - name: config-volume
           mountPath: "/etc/debezium-connector-config"
+      - name: controller
+        image: {{ include "gart.controller.image" . }}
+        imagePullPolicy: {{ .Values.controller.image.pullPolicy | quote }}
+        command: ["/workspace/gart/scripts/controller.py"]
+        ports:
+          - containerPort: {{ .Values.controller.containerPort }}
+        envFrom:
+        - configMapRef:
+            name: {{ include "gart.configmapName" . }}
       volumes:
       - name: config-volume
         configMap:

charts/gart/templates/converter/svc.yaml

@@ -0,0 +1,16 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{ include "gart.controller-service.fullname" . }}
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app: converter
+spec:
+  type: ClusterIP
+  selector:
+    app: converter
+  ports:
+    - protocol: TCP
+      port: {{ .Values.controller.port }}
+      targetPort: {{ .Values.controller.containerPort }}
+

charts/gart/templates/debezium/configmap.yaml

@@ -11,7 +11,7 @@ metadata:
 data:
   mysql-connector.json: |-
     {
-      "name": "debezium-connector-mysql",
+      "name": {{ include "gart.debezium.fullname" . | quote }},
       "config": {
         {{- if eq .Values.dataconfig.dbType "mysql" }}
         "connector.class": "io.debezium.connector.mysql.MySqlConnector",

charts/gart/values.yaml

@@ -155,12 +155,6 @@ debezium:
     pullPolicy: IfNotPresent
   containerPort: 8083
 
-curl:
-  image:
-    repository: curlimages/curl
-    tag: latest
-    pullPolicy: IfNotPresent
-
 # converter config
 converter:
   replicaCount: 1
@@ -184,6 +178,17 @@ analyzer:
     tag: latest
     pullPolicy: IfNotPresent
 
+# controller config
+controller:
+  image:
+    repository: gart-converter
+    tag: latest
+    pullPolicy: IfNotPresent
+  #service port
+  port: 80
+  # container port
+  containerPort: 5000
+
 dataconfig:
   subgraphNum: 2
   dbHost: "127.0.0.1"

scripts/controller.py

@@ -0,0 +1,36 @@
+#!/usr/bin/env python3
+
+from flask import Flask
+import subprocess
+import os
+
+app = Flask(__name__)
+port = int(os.getenv("CONTROLLER_FLASK_PORT", 5000))
+
+
+@app.route("/control/pause", methods=["POST"])
+def pause():
+    subprocess.run(
+        [
+            "/bin/bash",
+            "-c",
+            "/workspace/gart/scripts/pause_resume_data_processing.sh pause",
+        ]
+    )
+    return "Paused", 200
+
+
+@app.route("/control/resume", methods=["POST"])
+def resume():
+    subprocess.run(
+        [
+            "/bin/bash",
+            "-c",
+            "/workspace/gart/scripts/pause_resume_data_processing.sh resume",
+        ]
+    )
+    return "Resumed", 200
+
+
+if __name__ == "__main__":
+    app.run(host="0.0.0.0", port=port)

scripts/install-deps.sh

@@ -146,7 +146,7 @@ fi
 # for psycopg2, you need to install libpq-dev
 
 sudo apt-get install -y libpq-dev
-pip3 install sqlalchemy pymysql psycopg2 etcd3 libclang
+pip3 install sqlalchemy pymysql psycopg2 etcd3 libclang flask
 
 if [ "$ROLE" == "All" ]; then
   # Install requirements-dev.txt for docs

scripts/pause_resume_data_processing.sh

@@ -0,0 +1,37 @@
+#!/bin/bash
+
+op=$1
+
+check_env_vars() {
+    if [[ -z $DEBEZIUM_SERVICE ]] || [[ -z $DEBEZIUM_CONNECTOR_NAME ]]; then
+        echo "Debezium service URL or connector name not set."
+        exit 1
+    fi
+}
+
+if [[ $op == "pause" ]]; then
+    echo "Pausing data processing"
+    check_env_vars
+    response=$(curl -s -o /dev/null -w "%{http_code}" -X PUT "http://${DEBEZIUM_SERVICE}/connectors/${DEBEZIUM_CONNECTOR_NAME}/pause")
+    if [[ $response != 202 ]]; then
+        echo "Failed to pause connector"
+        exit 1
+    else
+        echo "Connector paused successfully"
+        exit 0
+    fi
+elif [[ $op == "resume" ]]; then
+    echo "Resuming data processing"
+    check_env_vars
+    response=$(curl -s -o /dev/null -w "%{http_code}" -X PUT "http://${DEBEZIUM_SERVICE}/connectors/${DEBEZIUM_CONNECTOR_NAME}/resume")
+    if [[ $response != 202 ]]; then
+        echo "Failed to resume connector"
+        exit 1
+    else
+        echo "Connector resumed successfully"
+        exit 0
+    fi
+else
+    echo "Invalid operation $op"
+    exit 1
+fi