Add user course points realtime consumer; scale batch consumer down (#…

…1185)

* add user course points realtime consumer; scale batch consumer down

* fix helm chart config key

backend/bin/kafkaConsumer/userCoursePointsConsumer/interfaces.ts → backend/bin/kafkaConsumer/common/userCoursePoints/interfaces.ts

@@ -1,4 +1,4 @@
-import { Message as UserPointsMessage } from "../common/userPoints/interfaces"
+import { Message as UserPointsMessage } from "../userPoints/interfaces"
 
 export interface Message {
   timestamp: string

backend/bin/kafkaConsumer/userCoursePointsConsumer/saveToDB.ts → backend/bin/kafkaConsumer/common/userCoursePoints/saveToDB.ts

@@ -1,13 +1,13 @@
-import { ok } from "../../../util/result"
-import { KafkaContext } from "../common/kafkaContext"
-import { checkCompletion } from "../common/userFunctions"
+import { ok } from "../../../../util/result"
+import { KafkaContext } from "../kafkaContext"
+import { checkCompletion } from "../userFunctions"
 import {
   createExerciseCompletion,
   getCreatedAndUpdatedExerciseCompletions,
   pruneExerciseCompletions,
   updateExerciseCompletion,
-} from "../common/userPoints/exerciseCompletionFunctions"
-import { getCourse, getTimestamp, getUser } from "../common/userPoints/util"
+} from "../userPoints/exerciseCompletionFunctions"
+import { getCourse, getTimestamp, getUser } from "../userPoints/util"
 import { Message } from "./interfaces"
 
 export const saveToDatabase = async (

backend/bin/kafkaConsumer/userCoursePointsConsumer/validate.ts → backend/bin/kafkaConsumer/common/userCoursePoints/validate.ts

@@ -1,6 +1,6 @@
 import * as yup from "yup"
 
-import { MessageYupSchema as UserPointsMessageYupSchema } from "../common/userPoints/validate"
+import { MessageYupSchema as UserPointsMessageYupSchema } from "../userPoints/validate"
 
 const CURRENT_MESSAGE_FORMAT_VERSION = 1
 

backend/bin/kafkaConsumer/kafkaConfig.ts

@@ -17,6 +17,9 @@ const kafkaConfig = {
   user_course_points_consumer: {
     topic_name: "user-course-points-batch",
   },
+  user_course_points_realtime_consumer: {
+    topic_name: "user-course-points-realtime",
+  },
   commit_interval: 50,
 }
 

backend/bin/kafkaConsumer/userCoursePointsConsumer/kafkaConsumer.ts

@@ -9,9 +9,9 @@ import { createKafkaConsumer } from "../common/createKafkaConsumer"
 import { handleMessage } from "../common/handleMessage"
 import { KafkaContext } from "../common/kafkaContext"
 import { handledRecently, setHandledRecently } from "../common/messageHashCache"
+import { saveToDatabase } from "../common/userCoursePoints/saveToDB"
+import { MessageYupSchema } from "../common/userCoursePoints/validate"
 import config from "../kafkaConfig"
-import { saveToDatabase } from "./saveToDB"
-import { MessageYupSchema } from "./validate"
 
 const TOPIC_NAME = [config.user_course_points_consumer.topic_name]
 

backend/bin/kafkaConsumer/userCoursePointsRealtimeConsumer/kafkaConsumer.ts

@@ -0,0 +1,67 @@
+import { Message as KafkaMessage, LibrdKafkaError } from "node-rdkafka"
+
+import { Mutex } from "../../../lib/await-semaphore"
+import { KafkaError } from "../../../lib/errors"
+import sentryLogger from "../../../lib/logger"
+import prisma from "../../../prisma"
+import knex from "../../../services/knex"
+import { createKafkaConsumer } from "../common/createKafkaConsumer"
+import { handleMessage } from "../common/handleMessage"
+import { KafkaContext } from "../common/kafkaContext"
+import { Message } from "../common/userCoursePoints/interfaces"
+import { saveToDatabase } from "../common/userCoursePoints/saveToDB"
+import { MessageYupSchema } from "../common/userCoursePoints/validate"
+import config from "../kafkaConfig"
+
+const TOPIC_NAME = [config.user_course_points_realtime_consumer.topic_name]
+
+const mutex = new Mutex()
+
+const logger = sentryLogger({
+  service: "kafka-consumer-user-course-points-realtime",
+})
+const consumer = createKafkaConsumer({ logger, prisma })
+
+consumer.connect()
+
+const context: KafkaContext = {
+  prisma,
+  logger,
+  mutex,
+  consumer,
+  knex,
+  topic_name: TOPIC_NAME[0],
+}
+
+consumer.on("ready", () => {
+  logger.info("Ready to consume")
+  consumer.subscribe(TOPIC_NAME)
+
+  const consumerImpl = async (
+    error: LibrdKafkaError,
+    messages: KafkaMessage[],
+  ) => {
+    if (error) {
+      logger.error(new KafkaError("Error while consuming", error))
+      process.exit(-1)
+    }
+
+    if (messages.length > 0) {
+      await handleMessage<Message>({
+        context,
+        kafkaMessage: messages[0],
+        MessageYupSchema,
+        saveToDatabase,
+      })
+      setImmediate(() => {
+        consumer.consume(1, consumerImpl)
+      })
+    } else {
+      setTimeout(() => {
+        consumer.consume(1, consumerImpl)
+      }, 10)
+    }
+  }
+
+  consumer.consume(1, consumerImpl)
+})

backend/package.json

@@ -26,6 +26,7 @@
     "kafka-consumer-user-points-batch": "node ./dist/bin/kafkaConsumer/userPointsConsumer/kafkaConsumer.js",
     "kafka-consumer-user-points-realtime": "node ./dist/bin/kafkaConsumer/userPointsRealtimeConsumer/kafkaConsumer.js",
     "kafka-consumer-user-course-points-batch": "node ./dist/bin/kafkaConsumer/userCoursePointsConsumer/kafkaConsumer.js",
+    "kafka-consumer-user-course-points-realtime": "node ./dist/bin/kafkaConsumer/userCoursePointsRealtimeConsumer/kafkaConsumer.js",
     "kafka-consumer-exercises": "node ./dist/bin/kafkaConsumer/exerciseConsumer/kafkaConsumer.js",
     "generate": "npm run generate:prisma && npm run generate:nexus",
     "generate:prisma": "prisma generate",

helm/templates/kafka/kafka-consumer-user-course-points-realtime.yml

@@ -0,0 +1,53 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: kafka-consumer-user-course-points-realtime
+  labels:
+    {{- include "helm.labels" . | nindent 4 }}
+spec:
+  selector:
+    matchLabels:
+      app: kafka-consumer-user-course-points-realtime
+      {{- include "helm.selectorLabels" . | nindent 6 }}
+  replicas: {{ .Values.kafkaConsumer.userCoursePointsRealtime.replicaCount }}
+  template:
+    metadata:
+      labels:
+        app: kafka-consumer-user-course-points-realtime
+        {{- include "helm.selectorLabels" . | nindent 8 }}
+    spec:
+      containers:
+        - name: kafka-consumer-user-course-points-realtime
+          image: "{{ .Values.image.repository }}/moocfi-backend:{{ .Values.image.tag | default .Chart.AppVersion }}"
+          command: ["sh", "-c", "npm run kafka-consumer-user-course-points-realtime"]
+          imagePullPolicy: Always
+          ports:
+            - name: backend-http
+              containerPort: 4000
+          resources:
+            limits:
+              memory: 800Mi
+              cpu: 80m
+            requests:
+              memory: 200Mi
+              cpu: 20m
+          envFrom:
+          - secretRef:
+              name: backend-secret
+          env:
+            - name: REDIS_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: redis
+                  key: redis-password
+          volumeMounts:
+            - name: google-cloud-storage-serviceaccount
+              mountPath: "/etc/gcs"
+              readOnly: true
+      volumes:
+        - name: google-cloud-storage-serviceaccount
+          secret:
+            secretName: google-cloud-storage-serviceaccount
+            items:
+              - key: account.json
+                path: account.json

helm/templates/kafka/user-course-points-realtime-topic.yml

@@ -0,0 +1,13 @@
+apiVersion: kafka.strimzi.io/v1beta1
+kind: KafkaTopic
+metadata:
+  name: user-course-points-realtime
+  namespace: kafka
+  labels:
+    strimzi.io/cluster: kafka-cluster
+spec:
+  partitions: 10
+  replicas: 2
+  config:
+    retention.ms: 604800000
+    segment.bytes: 107374182

helm/values.yaml

@@ -23,7 +23,9 @@ kafkaConsumer:
   userPointsRealtime:
     replicaCount: 1
   userCoursePoints:
-    replicaCount: 6
+    replicaCount: 2
+  userCoursePointsRealtime:
+    replicaCount: 1
 
 shibboTest:
   replicaCount: 0