index.js

// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.

require('module-alias/register');
require('dotenv').config();
const AsyncLock = require('async-lock');
const _ = require('lodash');
const DatabaseModel = require('openpaidbsdk');
const { default: PQueue } = require('p-queue');
const interval = require('interval-promise');
require('@dbc/common/init');
const logger = require('@dbc/common/logger');
const { getEventInformer } = require('@dbc/common/k8s');
const { alwaysRetryDecorator } = require('@dbc/common/util');
const disk = require('diskusage');
const config = require('@dbc/watcher/cluster-event/config');

// Here, we use AsyncLock to control the concurrency of events with the same uid;
// e.g. If one event has ADDED, MODIFED, and MODIFED incidents, we use AsyncLock
// to ensure they will be delivered to write-merger in order.
// In the same time, we use PQueue to control the concurrency of events with different uid;
// e.g. If there are event 1 ~ event 30000, only some of them can be processed concurrently.
const lock = new AsyncLock({ maxPending: Number.MAX_SAFE_INTEGER });
const queue = new PQueue({ concurrency: config.maxRpcConcurrency });
const databaseModel = new DatabaseModel(
  config.dbConnectionStr,
  config.maxDatabaseConnection,
);

async function synchronizeEvent(eventType, apiObject) {
  // query db instead
  const uid = apiObject.metadata.uid;
  const names = apiObject.involvedObject.name.split('-');

  const obj = {
    uid: uid,
    frameworkName: names[0],
    podUid: apiObject.involvedObject.uid,
    taskroleName: names[1],
    taskName: apiObject.involvedObject.name,
    taskIndex: parseInt(names[2]),
    type: apiObject.type,
    reason: apiObject.reason,
    message: apiObject.message,
    firstTimestamp: apiObject.firstTimestamp,
    lastTimestamp: apiObject.lastTimestamp,
    count: apiObject.count,
    sourceComponent: _.get(apiObject, 'source.component', null),
    sourceHost: _.get(apiObject, 'source.host', null),
    event: JSON.stringify(apiObject),
  };

  databaseModel.FrameworkEvent.upsert(obj, { where: { uid: uid } });
}

const eventHandler = (eventType, apiObject) => {
  /*
    framework name-based lock + always retry
  */
  const receivedTs = new Date().getTime();
  const involvedObjKind = apiObject.involvedObject.kind;
  const involvedObjName = apiObject.involvedObject.name;
  const uid = apiObject.metadata.uid;
  if (
    involvedObjKind === 'Pod' &&
    /^[a-z0-9]{32}-[A-Za-z0-9._~]+-[0-9]+$/.test(involvedObjName)
  ) {
    logger.info(
      `Cluster event type=${eventType} receivedTs=${receivedTs} uid=${uid} involvedObjKind=${involvedObjKind} involvedObjName=${involvedObjName} received.`,
    );
    lock.acquire(uid, () => {
      return queue.add(
        alwaysRetryDecorator(
          () => synchronizeEvent(eventType, apiObject),
          `Sync to database type=${eventType} receivedTs=${receivedTs} uid=${uid} involvedObjKind=${involvedObjKind} involvedObjName=${involvedObjName}`,
        ),
      );
    });
  } else {
    logger.info(
      `Cluster Event type=${eventType} receivedTs=${receivedTs} uid=${uid} involvedObjKind=${involvedObjKind} involvedObjName=${involvedObjName} received but ignored.`,
    );
  }
};

async function assertDiskUsageHealthy() {
  try {
    const { available, total } = await disk.check(config.diskPath);
    const currentUsage = ((total - available) / total) * 100;
    logger.info(`Current internal storage usage is ${currentUsage}% .`);
    if (currentUsage > config.maxDiskUsagePercent) {
      logger.error(
        `Internal storage usage exceeds ${config.maxDiskUsagePercent}%, exit.`,
        function() {
          process.exit(1);
        },
      );
    }
  } catch (err) {
    logger.error(`Check disk usage fails, details: ${err}`, function() {
      process.exit(1);
    });
  }
}

function startInformer() {
  const informer = getEventInformer();

  informer.on('add', apiObject => {
    eventHandler('ADDED', apiObject);
  });
  informer.on('update', apiObject => {
    eventHandler('MODIFED', apiObject);
  });
  informer.on('delete', apiObject => {
    eventHandler('DELETED', apiObject);
  });
  informer.on('error', err => {
    // If any error happens, the process should exit, and let Kubernetes restart it.
    logger.error(err, function() {
      process.exit(1);
    });
  });
  informer.start();
}

function startDiskCheck() {
  interval(assertDiskUsageHealthy, config.diskCheckIntervalSecond * 1000, {
    stopOnError: false,
  });
}

async function main() {
  await assertDiskUsageHealthy();
  startInformer();
  startDiskCheck();
}

main();