Test running updater in CI

.github/workflows/release-sis-updater-scheduler.yaml

@@ -1,54 +0,0 @@
-name: "Release sis-updater-scheduler"
-
-on:
-  push:
-    branches:
-      - master
-    paths:
-      - '.github/workflows/release-sis-updater-scheduler.yaml'
-      - 'updater/sis-updater-scheduler/**'
-
-jobs:
-  build_and_release:
-    name: "Build and release sis-updater-scheduler"
-    runs-on: ubuntu-latest
-    steps:
-    - uses: actions/checkout@v2
-    - uses: docker/setup-buildx-action@v1
-    - name: "Login to DockerHub"
-      uses: docker/login-action@v1
-      with:
-        username: ${{ secrets.DOCKERHUB_USERNAME }}
-        password: ${{ secrets.DOCKERHUB_TOKEN }}
-    - name: "Build and push"
-      uses: docker/build-push-action@v2
-      with:
-        cache-from: type=registry,ref=toska/sis-updater-scheduler:latest
-        cache-to: type=inline
-        context: updater/sis-updater-scheduler/
-        push: true
-        tags: toska/sis-updater-scheduler:latest
-        target: production
-    - name: "Send notification to slack"
-      uses:  UniversityOfHelsinkiCS/slackbot-action@v1.4.2
-      with:
-        webhook-url: ${{ secrets.WEBHOOK_URL }}
-        message-type: deployment
-        softa-url: https://oodikone.cs.helsinki.fi
-        deployment-target: production
-        softa-to-deploy: sis-updater-scheduler
-
-  report_failure:
-    runs-on: ubuntu-latest
-    name: "Report deployment failure to oodikone notifications"
-    needs:
-    - build_and_release
-    if: failure()
-    steps:
-      - name: "Send notification to slack"
-        uses:  UniversityOfHelsinkiCS/slackbot-action@v1.4.2
-        with:
-          webhook-url: ${{ secrets.WEBHOOK_URL }}
-          message-type: deployment-failure
-          deployment-target: production
-          softa-to-deploy: sis-updater-scheduler

.github/workflows/release-sis-updater-worker.yaml

@@ -1,54 +0,0 @@
-name: "Release sis-updater-worker"
-
-on:
-  push:
-    branches:
-      - master
-    paths:
-      - '.github/workflows/release-sis-updater-worker.yaml'
-      - 'updater/sis-updater-worker/**'
-
-jobs:
-  build_and_release:
-    name: "Build and release sis-updater-worker"
-    runs-on: ubuntu-latest
-    steps:
-    - uses: actions/checkout@v2
-    - uses: docker/setup-buildx-action@v1
-    - name: "Login to DockerHub"
-      uses: docker/login-action@v1
-      with:
-        username: ${{ secrets.DOCKERHUB_USERNAME }}
-        password: ${{ secrets.DOCKERHUB_TOKEN }}
-    - name: "Build and push"
-      uses: docker/build-push-action@v2
-      with:
-        cache-from: type=registry,ref=toska/sis-updater-worker:latest
-        cache-to: type=inline
-        context: updater/sis-updater-worker/
-        push: true
-        tags: toska/sis-updater-worker:latest
-        target: production
-    - name: "Send notification to slack"
-      uses:  UniversityOfHelsinkiCS/slackbot-action@v1.4.2
-      with:
-        webhook-url: ${{ secrets.WEBHOOK_URL }}
-        message-type: deployment
-        softa-url: https://oodikone.cs.helsinki.fi
-        deployment-target: production
-        softa-to-deploy: sis-updater-worker
-
-  report_failure:
-    runs-on: ubuntu-latest
-    name: "Report deployment failure to oodikone notifications"
-    needs:
-    - build_and_release
-    if: failure()
-    steps:
-      - name: "Send notification to slack"
-        uses:  UniversityOfHelsinkiCS/slackbot-action@v1.4.2
-        with:
-          webhook-url: ${{ secrets.WEBHOOK_URL }}
-          message-type: deployment-failure
-          deployment-target: production
-          softa-to-deploy: sis-updater-worker

.github/workflows/test-and-release-updater.yaml

@@ -0,0 +1,131 @@
+name: 'Test and release updater'
+
+on:
+  push:
+    branches:
+      - master
+    paths:
+      - '.github/workflows/test-and-release-updater.yaml'
+      - 'updater/sis-updater-worker/**'
+      - 'updater/sis-updater-scheduler/**'
+
+jobs:
+  build_worker:
+    name: 'Build sis-updater-worker'
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - uses: docker/setup-buildx-action@v1
+      - name: 'Build the image'
+        uses: docker/build-push-action@v2
+        with:
+          context: updater/sis-updater-worker/
+          target: production
+          push: false
+          outputs: type=docker,dest=/tmp/worker.tar
+          tags: toska/sis-updater-worker:latest
+          cache-from: type=registry,ref=toska/sis-updater-worker:latest
+          cache-to: type=inline
+      - name: 'Store built image'
+        uses: actions/artifact@v2
+        with:
+          path: /tmp/worker.tar
+
+  build_scheduler:
+    name: 'Build sis-updater-scheduler'
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - uses: docker/setup-buildx-action@v1
+      - name: 'Build the image'
+        uses: docker/build-push-action@v2
+        with:
+          context: updater/sis-updater-scheduler/
+          target: production
+          push: false
+          outputs: type=docker,dest=/tmp/scheduler.tar
+          tags: toska/sis-updater-scheduler:latest
+          cache-from: type=registry,ref=toska/sis-updater-scheduler:latest
+          cache-to: type=inline
+      - name: 'Store built image'
+        uses: actions/artifact@v2
+        with:
+          path: /tmp/scheduler.tar
+
+  test_updater:
+    name: 'Test updater'
+    runs-on: ubuntu-latest
+    needs:
+      - build_worker
+      - build_scheduler
+    steps:
+      - name: 'Download built images'
+        uses: actions/download-artifact@v2
+        with:
+          path: /tmp/
+      - name: 'Load worker image'
+        run: docker load --input /tmp/artifact/worker.tar
+      - name: 'Load scheduler image'
+        run: docker load --input /tmp/artifact/scheduler.tar
+      - name: 'Start worker'
+        run: docker-compose up --detach --no-recreate --file docker-compose.ci.yml sis-updater-worker
+      - name: 'Run hourly jobs'
+        run: docker-compose run sis-updater-scheduler npm start
+        env:
+          SCHEDULE_IMMEDIATE: hourly
+          EXIT_AFTER_IMMEDIATES: yes
+      - name: 'Run weekly jobs'
+        run: docker-compose run sis-updater-scheduler npm start
+        env:
+          SCHEDULE_IMMEDIATE: weekly
+          EXIT_AFTER_IMMEDIATES: yes
+      - name: 'Run prepurge'
+        run: docker-compose run sis-updater-scheduler npm start
+        env:
+          SCHEDULE_IMMEDIATE: prepurge
+          EXIT_AFTER_IMMEDIATES: yes
+      - name: 'Run purge'
+        run: docker-compose run sis-updater-scheduler npm start
+        env:
+          SCHEDULE_IMMEDIATE: purge
+          EXIT_AFTER_IMMEDIATES: yes
+
+  release_updater:
+    name: 'Release updater'
+    runs-on: ubuntu-latest
+    needs:
+      - test_updater
+    steps:
+      - uses: actions/checkout@v2
+      - uses: docker/setup-buildx-action@v1
+      - name: 'Login to DockerHub'
+        uses: docker/login-action@v1
+        with:
+          username: ${{ secrets.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+      - name: 'Build and push scheduler'
+        uses: docker/build-push-action@v2
+        with:
+          cache-from: type=registry,ref=toska/sis-updater-scheduler:latest
+          cache-to: type=inline
+          context: updater/sis-updater-scheduler/
+          push: true
+          tags: toska/sis-updater-scheduler:latest
+          target: production
+      - name: 'Build and push worker'
+        uses: docker/build-push-action@v2
+        with:
+          cache-from: type=registry,ref=toska/sis-updater-worker:latest
+          cache-to: type=inline
+          context: updater/sis-updater-worker/
+          push: true
+          tags: toska/sis-updater-worker:latest
+          target: production
+      - name: 'Send notification to slack'
+        uses: UniversityOfHelsinkiCS/slackbot-action@v1.4.2
+        with:
+          webhook-url: ${{ secrets.WEBHOOK_URL }}
+          message-type: deployment
+          softa-url: https://oodikone.cs.helsinki.fi
+          deployment-target: production
+          softa-to-deploy: sis-updater

docker-compose.ci.yml

@@ -103,9 +103,35 @@ services:
     profiles:
       - updater
 
+  sis-updater-scheduler:
+    <<: *defaults
+    build:
+      target: development
+      context: updater/sis-updater-scheduler
+    container_name: sis-updater-scheduler
+    depends_on:
+      - redis
+      - sis-importer-db
+      - sis-updater-nats
+    environment:
+      REDIS_URI: //redis:6379
+      SECRET_TOKEN: dev
+      SIS_IMPORTER_DATABASE: sis-importer-db
+      SIS_IMPORTER_HOST: sis-importer-db
+      SIS_IMPORTER_PASSWORD: postgres
+      SIS_IMPORTER_USER: postgres
+      SIS_NATS_TOKEN: dev
+      SIS_NATS_URI: sis-updater-nats://sis-updater-nats:4222
+      SCHEDULE_IMMEDIATE: '${SCHEDULE_IMMEDIATE}'
+      EXIT_AFTER_IMMEDIATES: '${EXIT_AFTER_IMMEDIATES}'
+    ports:
+      - 8082:8082
+    profiles:
+      - updater
+
   sis-updater-nats:
     <<: *defaults
-    command: '-cid sis-updater-nats -m 8222 -p 4222 --max_age 2h --hb_fail_count 2 --hb_timeout 10s --hb_interval 10s --auth dev --file_slice_max_bytes 0 --file_slice_max_age 60s --max_msgs 0 -store file -dir datastore'
+    command: '-cid sis-updater-nats -m 8222 -p 4222 --max_channels 1000 --max_age 2h --hb_fail_count 2 --hb_timeout 10s --hb_interval 10s --auth dev --file_slice_max_bytes 0 --file_slice_max_age 60s --max_msgs 0 -store file -dir datastore'
     container_name: sis-updater-nats
     image: nats-streaming:0.16.2
     profiles:

docker-compose.yml

@@ -1,3 +1,5 @@
+version: '2.1'
+
 x-defaults: &defaults
   restart: unless-stopped
 
@@ -97,7 +99,7 @@ services:
 
   sis-updater-nats:
     <<: *defaults
-    command: '-cid sis-updater-nats -m 8222 -p 4222 --max_age 2h --hb_fail_count 2 --hb_timeout 10s --hb_interval 10s --auth dev --file_slice_max_bytes 0 --file_slice_max_age 60s --max_msgs 0 -store file -dir datastore'
+    command: '-cid sis-updater-nats -m 8222 -p 4222 --max_channels 1000 --max_age 2h --hb_fail_count 2 --hb_timeout 10s --hb_interval 10s --auth dev --file_slice_max_bytes 0 --file_slice_max_age 60s --max_msgs 0 -store file -dir datastore'
     container_name: sis-updater-nats
     image: nats-streaming:0.16.2
     profiles:
@@ -122,6 +124,8 @@ services:
       SIS_IMPORTER_USER: postgres
       SIS_NATS_TOKEN: dev
       SIS_NATS_URI: sis-updater-nats://sis-updater-nats:4222
+      SCHEDULE_IMMEDIATE: '${SCHEDULE_IMMEDIATE}'
+      EXIT_AFTER_IMMEDIATES: '${EXIT_AFTER_IMMEDIATES}'
     ports:
       - 8082:8082
     profiles:

updater/sis-updater-scheduler/src/index.js

@@ -22,17 +22,77 @@ knexConnection.on('error', e => {
   if (!process.env.CI) process.exit(1)
 })
 
+const JOB_TYPES = {
+  hourly: scheduleHourly,
+  weekly: scheduleWeekly,
+  prepurge: schedulePrePurge,
+  purge: schedulePurge,
+}
+
+const scheduleJob = async type => {
+  if (!JOB_TYPES[type]) {
+    logger.error({
+      message: `Cannot schedule unknown job type '${type}'.`,
+      job: type,
+    })
+
+    return
+  }
+
+  logger.info(`Scheduling job type '${type}'.`)
+
+  await JOB_TYPES[type]()
+
+  logger.info(`Job '${type}' finished.`)
+}
+
+const runImmediates = async () => {
+  const jobs = (process.env.SCHEDULE_IMMEDIATE || 'none').split(',')
+  await Promise.all(jobs.map(scheduleJob))
+}
+
+const handleImmediates = async () => {
+  const exit = process.env.EXIT_AFTER_IMMEDIATES === 'yes'
+
+  try {
+    await runImmediates()
+
+    if (exit) {
+      process.exit(0)
+    }
+  } catch (e) {
+    logger.error({
+      message: `Running immediate job failed: ${e.message}`,
+      meta: e.stack,
+    })
+
+    if (exit) {
+      process.exit(1)
+    }
+  }
+}
+
 knexConnection.on('connect', async () => {
   logger.info('Knex database connection established successfully')
   startServer()
 
+  await handleImmediates()
+
   // Monday-Friday at every minute 30
   scheduleCron('30 * * * 1-5', async () => {
     // If updater is currently running, then return
     if ((await isUpdaterActive()) || isDev) return
     logger.info('Starting hourly')
 
-    await scheduleHourly()
+    try {
+      await scheduleHourly()
+    } catch (e) {
+      logger.error({
+        message: 'Hourly run failed: ' + e.message,
+        meta: e.stack,
+      })
+    }
+
     await redisSet(REDIS_LAST_HOURLY_SCHEDULE, new Date())
   })
 
@@ -41,7 +101,15 @@ knexConnection.on('connect', async () => {
     if (isDev) return
     logger.info('Starting weekly')
 
-    await scheduleWeekly()
+    try {
+      await scheduleWeekly()
+    } catch (e) {
+      logger.error({
+        message: 'Weekly run failed: ' + e.message,
+        meta: e.stack,
+      })
+    }
+
     await redisSet(REDIS_LAST_WEEKLY_SCHEDULE, new Date())
   })
 
@@ -49,7 +117,15 @@ knexConnection.on('connect', async () => {
   scheduleCron('0 12 * * MON', async () => {
     if (isDev) return
     logger.info('Starting prepurge')
-    await schedulePrePurge()
+
+    try {
+      await schedulePrePurge()
+    } catch (e) {
+      logger.error({
+        message: 'Prepurge failed: ' + e.message,
+        meta: e.stack,
+      })
+    }
   })
 
   // Sunday at 4 AM
@@ -57,6 +133,14 @@ knexConnection.on('connect', async () => {
     // If updater is currently running, then return
     if ((await isUpdaterActive()) || isDev) return
     logger.info('Starting purge')
-    await schedulePurge()
+
+    try {
+      await schedulePurge()
+    } catch (e) {
+      logger.error({
+        message: 'Purge failed: ' + e.message,
+        meta: e.stack,
+      })
+    }
   })
 })

updater/sis-updater-scheduler/src/scheduler.js

@@ -1,9 +1,11 @@
 const { chunk } = require('lodash')
 const { eachLimit } = require('async')
+const uuid = require('uuid')
 const { knexConnection } = require('./db/connection')
-const { stan } = require('./utils/stan')
+const { stan, opts } = require('./utils/stan')
 const { incrby: redisIncrementBy, get: redisGet } = require('./utils/redis')
 const {
+  NATS_GROUP,
   SIS_UPDATER_SCHEDULE_CHANNEL,
   SIS_MISC_SCHEDULE_CHANNEL,
   CHUNK_SIZE,
@@ -36,11 +38,43 @@ const createJobs = async (entityIds, type, channel = SIS_UPDATER_SCHEDULE_CHANNE
   const redisKey = type === 'students' ? REDIS_TOTAL_STUDENTS_KEY : REDIS_TOTAL_META_KEY
   await redisIncrementBy(redisKey, entityIds.length)
 
-  return new Promise((res, rej) => {
-    stan.publish(channel, JSON.stringify({ entityIds, type }), err => {
-      if (err) return rej(err)
-      res()
-    })
+  const id = uuid.v4()
+
+  const completionChannel = stan.subscribe('SIS_COMPLETED_CHANNEL-' + id, NATS_GROUP, opts)
+
+  return new Promise((resolve, reject) => {
+    const messageHandler = msg => {
+      let data = null
+
+      try {
+        data = JSON.parse(msg.getData())
+      } catch (err) {
+        logger.error({
+          message: 'Unable to parse completion message.',
+          meta: err.stack,
+        })
+
+        return
+      }
+
+      if (data.id === id) {
+        if (data.success) {
+          resolve()
+        } else {
+          logger.error('Job failed: ' + data.message)
+          reject(data.message)
+        }
+
+        completionChannel.off('message', messageHandler)
+        completionChannel.unsubscribe()
+      }
+
+      msg.ack()
+    }
+
+    completionChannel.on('message', messageHandler)
+
+    stan.publish(channel, JSON.stringify({ id, entityIds, type }))
   })
 }
 
@@ -195,6 +229,7 @@ const scheduleHourly = async () => {
     await eachLimit(chunk(personsToUpdate, CHUNK_SIZE), 10, async s => await createJobs(s, 'students'))
   } catch (e) {
     logger.error({ message: 'Hourly scheduling failed', meta: e.stack })
+    throw e
   }
 }
 
@@ -209,6 +244,7 @@ const scheduleProgrammes = async () => {
     await createJobs(entityIds, 'programme_modules')
   } catch (e) {
     logger.error({ message: 'Programme module scheduling failed', meta: e.stack })
+    throw e
   }
 }
 
@@ -218,6 +254,7 @@ const scheduleWeekly = async () => {
     await scheduleStudents()
   } catch (e) {
     logger.error({ message: 'Weekly scheduling failed', meta: e.stack })
+    throw e
   }
 }
 
@@ -226,6 +263,7 @@ const schedulePrePurge = async () => {
     await startPrePurge()
   } catch (e) {
     logger.error({ message: 'Purge failed', meta: e.stack })
+    throw e
   }
 }
 
@@ -234,6 +272,7 @@ const schedulePurge = async () => {
     await startPurge()
   } catch (e) {
     logger.error({ message: 'Purge failed', meta: e.stack })
+    throw e
   }
 }
 

updater/sis-updater-worker/src/index.js

@@ -19,9 +19,33 @@ const {
 } = require('./config')
 
 const handleMessage = messageHandler => async msg => {
+  let data = null
+
   try {
-    await messageHandler(JSON.parse(msg.getData()))
+    data = JSON.parse(msg.getData())
+  } catch (err) {
+    logger.error({
+      message: 'Failed to parse message',
+      meta: err.stack,
+    })
+  }
+
+  try {
+    await messageHandler(data)
+
+    if (data.id) {
+      logger.info('Completion Ack ' + data.id + ' (Success)')
+      stan.publish('SIS_COMPLETED_CHANNEL-' + data.id, JSON.stringify({ id: data.id, success: true }))
+    }
   } catch (e) {
+    if (data.id) {
+      logger.info('Completion Ack ' + data.id + ' (Failure)')
+      stan.publish(
+        'SIS_COMPLETED_CHANNEL-' + data.id,
+        JSON.stringify({ id: data.id, success: false, message: e.message })
+      )
+    }
+
     logger.error({ message: 'Failed handling message', meta: e.stack })
   } finally {
     try {