Repository: hyperjumptech/monika (ts)

Issue:
Optimize Probe Update Mechanism
I propose a better mechanism for Monika to receive probe data from Symon. There are significant changes on both side. Here are the steps:

1. Every time a probe changes (`add/update/delete`), a record will be stored in the `probe_changes` table.
2. Monika will continue to poll to Symon to get probe data. But, instead of retrieving all the probes, it will only get probe change data from a specific time.
3. When there are probe changes, Monika will take the following actions:

- `add` changes: Add the probe to the scheduler.
- `remove` changes: Remove the probe from the scheduler.
- `update` changes: Remove the probe from the scheduler and then re-add it.

The requirement for this mechanism to work is that Monika needs to change its mechanism to use a scheduler that is independent for each probe.

## Diagram
![probe-getter](https://github.com/hyperjumptech/monika/assets/15191978/2b987128-e121-4db1-8f06-7504026a7510)


Pull request:
Refactor: Separate Probe and Config Data
# Monika Pull Request (PR)

## What feature/issue does this PR add

Separate probe and config data to make it easy to update a probe independently. It is an effort to resolve #1163.

## How did you implement / how did you fix it

- refactor: remove dead code
- refactor: move config version state to context
- refactor: reuse update config function
- refactor: separate probe and config state
- feat: add symonGetProbesIntervalMs flag to make it easy to test
- feat: add stop method to SymonClient class to stop background process
- test: add test to simulate probe changes from Symon

## How to test

1. Run Monika, change the configuration.
2. Run Symon, update a probe.

## Video

### Monika

https://github.com/hyperjumptech/monika/assets/15191978/7ffd23e0-b061-4ba3-ab59-7cbd68f50f1c

### Symon

https://github.com/hyperjumptech/monika/assets/15191978/008aa8ba-5926-4b9e-9d01-3413b7400e35


Patch:
diff --git a/src/commands/monika.ts b/src/commands/monika.ts
index 2f0c405b..8c76f721 100644
--- a/src/commands/monika.ts
+++ b/src/commands/monika.ts
@@ -45,6 +45,7 @@ import {
   retryInitialDelayMs,
   retryMaxDelayMs,
   symonAPIVersion,
+  symonGetProbesIntervalMs,
 } from '../flag'
 import { printSummary, savePidFile } from '../jobs/summary-notification'
 import initLoaders from '../loaders'
@@ -222,6 +223,8 @@ export default class Monika extends Command {
       description: 'API Key for Symon',
     }),
 
+    symonGetProbesIntervalMs,
+
     symonLocationId: Flags.string({
       dependsOn: ['symonKey', 'symonUrl'],
       description: 'Location ID for Symon (optional)',
@@ -440,8 +443,8 @@ process.on('SIGINT', async () => {
   }
 
   if (symonClient) {
-    await symonClient.stopReport()
     await symonClient.sendStatus({ isOnline: false })
+    await symonClient.stop()
   }
 
   em.emit(events.application.terminated)
diff --git a/src/components/config/index.ts b/src/components/config/index.ts
index 113093f4..0e297a37 100644
--- a/src/components/config/index.ts
+++ b/src/components/config/index.ts
@@ -47,6 +47,7 @@ import {
   getConfigFrom,
   mergeConfigs,
 } from './get'
+import { getProbes, setProbes } from './probe'
 
 type ScheduleRemoteConfigFetcherParams = {
   configType: ConfigType
@@ -79,47 +80,41 @@ export const getConfig = (): Config => {
     }
   }
 
-  return config
+  return { ...config, probes: getProbes() }
 }
 
-export const updateConfig = async (
-  config: Config,
-  validate = true
-): Promise<void> => {
+export const updateConfig = async (config: Config): Promise<void> => {
   log.info('Updating config')
-  let validatedConfig = config
-  if (validate) {
-    try {
-      validatedConfig = await validateConfig(config)
-    } catch (error: any) {
-      if (isTestEnvironment) {
-        // return error during tests
-        throw new Error(error.message)
-      }
+  try {
+    const validatedConfig = await validateConfig(config)
+    const version = md5Hash(validatedConfig)
+    const hasChangeConfig = getContext().config?.version !== version
 
-      log.error(error?.message)
-      exit(1)
+    if (!hasChangeConfig) {
+      return
     }
-  }
-
-  const version = md5Hash(validatedConfig)
-  const hasChangeConfig = getContext()?.config?.version !== version
 
-  if (hasChangeConfig) {
     const newConfig = addConfigVersion(validatedConfig)
 
     setContext({ config: newConfig })
+    setProbes(newConfig.probes)
     emitter.emit(events.config.updated, newConfig)
-    log.warn('config file update detected')
+    log.info('Config file update detected')
+  } catch (error: any) {
+    if (isTestEnvironment) {
+      // return error during tests
+      throw new Error(error.message)
+    }
+
+    log.error(error?.message)
+    exit(1)
   }
 }
 
 export const setupConfig = async (flags: MonikaFlags): Promise<void> => {
   const validFlag = await createConfigIfEmpty(flags)
   const config = await getConfigFrom(validFlag)
-  const validatedConfig = await validateConfig(config)
-
-  setContext({ config: addConfigVersion(validatedConfig) })
+  await updateConfig(config)
 
   watchConfigsChange(validFlag)
 }
diff --git a/src/components/config/probe.ts b/src/components/config/probe.ts
new file mode 100644
index 00000000..b8db9986
--- /dev/null
+++ b/src/components/config/probe.ts
@@ -0,0 +1,43 @@
+import type { Probe } from '../../interfaces/probe'
+
+const probes: Map<string, Probe> = new Map()
+
+export function getProbes() {
+  const legacyProbes: Probe[] = []
+  for (const probe of probes.values()) {
+    legacyProbes.push(probe)
+  }
+
+  return legacyProbes
+}
+
+export function findProbe(id: string) {
+  return probes.get(id)
+}
+
+export function setProbes(newProbes: Probe[]) {
+  probes.clear()
+
+  for (const probe of newProbes) {
+    addProbe(probe)
+  }
+}
+
+export function addProbe(newProbe: Probe) {
+  return probes.set(newProbe.id, newProbe)
+}
+
+export function updateProbe(id: string, data: Probe): boolean {
+  const probe = findProbe(id)
+  if (!probe) {
+    return false
+  }
+
+  const updatedProbe = { ...probe, ...data }
+  probes.set(id, updatedProbe)
+  return true
+}
+
+export function deleteProbe(id: string) {
+  return probes.delete(id)
+}
diff --git a/src/context/index.ts b/src/context/index.ts
index d0964b6d..50eb1323 100644
--- a/src/context/index.ts
+++ b/src/context/index.ts
@@ -38,7 +38,7 @@ type Context = {
   // userAgent example: @hyperjumptech/monika/1.2.3 linux-x64 node-14.17.0
   userAgent: string
   incidents: Incident[]
-  config?: Config
+  config?: Omit<Config, 'probes'>
   flags: MonikaFlags
 }
 
diff --git a/src/flag.ts b/src/flag.ts
index ab2b78e5..cd574394 100644
--- a/src/flag.ts
+++ b/src/flag.ts
@@ -63,6 +63,7 @@ export type MonikaFlags = {
   symonMonikaId?: string
   symonReportInterval?: number
   symonReportLimit?: number
+  symonGetProbesIntervalMs: number
   symonUrl?: string
   text?: string
   ignoreInvalidTLS?: boolean
@@ -87,6 +88,10 @@ export const monikaFlagsDefaultValue: MonikaFlags = {
   // default is 20s interval lookup
   stun: 20,
   summary: false,
+  symonGetProbesIntervalMs: Number.parseInt(
+    process.env.FETCH_PROBES_INTERVAL ?? '60000',
+    10
+  ),
   verbose: false,
   version: undefined,
 }
@@ -98,6 +103,11 @@ export const symonAPIVersion = Flags.custom<SYMON_API_VERSION>({
   options: [SYMON_API_VERSION.v1, SYMON_API_VERSION.v2],
 })
 
+export const symonGetProbesIntervalMs = Flags.integer({
+  default: monikaFlagsDefaultValue.symonGetProbesIntervalMs,
+  description: `To determine how often Monika sends a request to Symon to get probe data, in milliseconds. Defaults to ${monikaFlagsDefaultValue.symonGetProbesIntervalMs}ms`,
+})
+
 export const retryInitialDelayMs = Flags.integer({
   default: monikaFlagsDefaultValue.retryInitialDelayMs,
   description: `The initial, first delay of the backoff retry when probe request is failed, in milliseconds. Defaults to ${monikaFlagsDefaultValue.retryInitialDelayMs}ms`,
diff --git a/src/symon/index.ts b/src/symon/index.ts
index f1cb13d4..538da79e 100644
--- a/src/symon/index.ts
+++ b/src/symon/index.ts
@@ -30,8 +30,8 @@ import { hostname } from 'os'
 import path from 'path'
 import Piscina from 'piscina'
 
+import { getProbes } from '../components/config/probe'
 import { updateConfig } from '../components/config'
-import { validateProbes } from '../components/config/validation'
 import { getOSName } from '../components/notification/alert-message'
 import { getContext } from '../context'
 import events from '../events'
@@ -65,18 +65,6 @@ type SymonHandshakeData = {
   version: string
 }
 
-type SymonClientEvent = {
-  alertId: string
-  event: 'incident' | 'recovery'
-  response: {
-    body?: unknown
-    headers?: Record<string, unknown>
-    size?: number
-    status: number // httpStatus Code
-    time?: number
-  }
-}
-
 type NotificationEvent = {
   probeID: string
   probeState: string
@@ -85,11 +73,16 @@ type NotificationEvent = {
   alertId: string
 }
 
-type ConfigListener = (config: Config) => void
-
-const isTestEnvironment = process.env.NODE_ENV === 'test' || process.env.CI
-
-let hasConnectionToSymon = false
+type SymonClientParams = Pick<
+  MonikaFlags,
+  | 'symon-api-version'
+  | 'symonKey'
+  | 'symonLocationId'
+  | 'symonMonikaId'
+  | 'symonReportInterval'
+  | 'symonReportLimit'
+  | 'symonUrl'
+>
 
 const getHandshakeData = async (): Promise<SymonHandshakeData> => {
   await retry(handleAll, {
@@ -98,11 +91,11 @@ const getHandshakeData = async (): Promise<SymonHandshakeData> => {
     await getPublicNetworkInfo()
       .then(({ city, hostname, isp, privateIp, publicIp }) => {
         log.info(
-          `Monika is running from: ${city} - ${isp} (${publicIp}) - ${hostname} (${privateIp})`
+          `[Symon] Monika is running from: ${city} - ${isp} (${publicIp}) - ${hostname} (${privateIp})`
         )
       })
       .catch((error) => {
-        log.error(`${error}. Retrying...`)
+        log.error(`[Symon] ${error}. Retrying...`)
         throw error
       })
   })
@@ -131,216 +124,181 @@ const getHandshakeData = async (): Promise<SymonHandshakeData> => {
   }
 }
 
-class SymonClient {
-  config: Config | null = null
-
-  configHash = ''
-
-  eventEmitter: EventEmitter | null = null
-
-  monikaId = ''
-
-  private apiKey = ''
-
-  private configListeners: ConfigListener[] = [] // (ms)
-
-  private fetchProbesInterval: number
-
+export default class SymonClient {
+  private readonly eventEmitter: EventEmitter = getEventEmitter()
+  private reportProbesInterval: number
+  private worker
+  private apiKey: string
+  private getProbesInterval: NodeJS.Timeout | undefined
+  private hasConnectionToSymon: boolean = false
   private httpClient: AxiosInstance
-
   private locationId: string
-
-  private probes: Probe[] = []
-
-  private reportProbesInterval = 10_000
-
+  private monikaId: string
   private reportProbesLimit: number
-
-  private url = ''
-
-  private worker = new Piscina.Piscina({
-    concurrentTasksPerWorker: 1,
-    // eslint-disable-next-line unicorn/prefer-module
-    filename: path.join(__dirname, '../../lib/workers/report-to-symon.js'),
-    idleTimeout: this.reportProbesInterval,
-    maxQueue: 1,
-  })
+  private reportTimeout: NodeJS.Timeout | undefined
+  private url: string
 
   constructor({
     'symon-api-version': apiVersion = SYMON_API_VERSION.v1,
     symonKey = '',
-    symonLocationId,
-    symonMonikaId,
-    symonReportInterval,
-    symonReportLimit,
+    symonLocationId = '',
+    symonMonikaId = '',
+    symonReportInterval = 10_000,
+    symonReportLimit = 100,
     symonUrl = '',
-  }: Pick<
-    MonikaFlags,
-    | 'symon-api-version'
-    | 'symonKey'
-    | 'symonLocationId'
-    | 'symonMonikaId'
-    | 'symonReportInterval'
-    | 'symonReportLimit'
-    | 'symonUrl'
-  >) {
+  }: SymonClientParams) {
+    this.apiKey = symonKey
+    this.url = symonUrl
     this.httpClient = axios.create({
-      baseURL: `${symonUrl}/api/${apiVersion}/monika`,
+      baseURL: `${this.url}/api/${apiVersion}/monika`,
       headers: {
-        'x-api-key': symonKey,
+        'x-api-key': this.apiKey,
       },
       timeout: DEFAULT_TIMEOUT,
     })
-
-    this.url = symonUrl
-
-    this.apiKey = symonKey
-
-    this.locationId = symonLocationId || ''
-
-    this.monikaId = symonMonikaId || ''
-
-    this.fetchProbesInterval = Number.parseInt(
-      process.env.FETCH_PROBES_INTERVAL ?? '60000',
-      10
-    )
-
-    this.reportProbesInterval = symonReportInterval ?? 10_000
-
-    this.reportProbesLimit = symonReportLimit ?? 100
+    this.locationId = symonLocationId
+    this.monikaId = symonMonikaId
+    this.reportProbesInterval = symonReportInterval
+    this.reportProbesLimit = symonReportLimit
+    this.worker = new Piscina.Piscina({
+      concurrentTasksPerWorker: 1,
+      // eslint-disable-next-line unicorn/prefer-module
+      filename: path.join(__dirname, '../../lib/workers/report-to-symon.js'),
+      idleTimeout: this.reportProbesInterval,
+      maxQueue: 1,
+    })
   }
 
   async initiate(): Promise<void> {
+    log.info('[Symon] Handshake starts')
     this.monikaId = await this.handshake()
-    await this.sendStatus({ isOnline: true })
+    log.info('[Symon] Handshake succeed')
+
+    log.info('[Symon] Send status')
+    this.sendStatus({ isOnline: true })
+      .then(() => {
+        log.info('[Symon] Send status succeed')
+      })
+      .catch((error) => {
+        log.error(`[Symon] Send status failed. ${(error as Error).message}`)
+      })
+
+    await this.fetchProbesAndUpdateConfig()
+    this.getProbesInterval = setInterval(
+      this.fetchProbesAndUpdateConfig.bind(this),
+      getContext().flags.symonGetProbesIntervalMs
+    )
 
-    log.debug('Handshake succesful')
+    this.report().catch((error) => {
+      this.hasConnectionToSymon = false
+      log.error(`[Symon] Report failed. ${(error as Error).message}`)
+    })
 
-    this.eventEmitter = getEventEmitter()
     this.eventEmitter.on(
       events.probe.notification.willSend,
-      ({ probeState, validation, alertId }: NotificationEvent) => {
-        this.willSendNotification(probeState, validation, alertId)
-      }
+      this.willSendEventListener.bind(this)
     )
+  }
 
-    await this.fetchProbesAndUpdateConfig()
-    if (!isTestEnvironment) {
-      setInterval(
-        this.fetchProbesAndUpdateConfig.bind(this),
-        this.fetchProbesInterval
-      )
+  async stop(): Promise<void> {
+    if (this.eventEmitter) {
+      this.eventEmitter.removeAllListeners()
     }
 
-    await this.report()
-    this.onConfig((config) => updateConfig(config, false))
-  }
+    clearInterval(this.getProbesInterval)
 
-  willSendNotification(
-    probeState: string,
-    validation: ValidatedResponse,
-    alertId: string
-  ): void {
-    this.notifyEvent({
-      alertId,
-      event: probeState === 'DOWN' ? 'incident' : 'recovery',
-      response: {
-        body: validation.response.data,
-        headers: validation.response.headers || {},
-        size: validation.response.headers['content-length'],
-        status: validation.response.status, // status is http status code
-        time: validation.response.responseTime,
-      },
-    }).catch((error: unknown) => {
-      log.error(error)
-    })
-  }
+    clearTimeout(this.reportTimeout)
 
-  async notifyEvent(event: SymonClientEvent): Promise<void> {
-    log.debug('Sending incident/recovery event to Symon')
-    await this.httpClient.post('/events', { monikaId: this.monikaId, ...event })
+    await this.worker.destroy()
   }
 
-  // monika subscribes to config update by providing listener callback
-  onConfig(listener: ConfigListener): unknown {
-    if (this.config) listener(this.config)
+  private willSendEventListener({
+    probeState,
+    validation,
+    alertId,
+  }: NotificationEvent) {
+    log.info(
+      `[Symon] Send ${
+        probeState === 'DOWN' ? 'incident' : 'recovery'
+      } event for Alert ID: ${alertId}`
+    )
 
-    this.configListeners.push(listener)
+    const { data, headers, responseTime, status } = validation.response
+    this.httpClient
+      .post('/events', {
+        monikaId: this.monikaId,
+        alertId,
+        event: probeState === 'DOWN' ? 'incident' : 'recovery',
+        response: {
+          body: data,
+          headers: typeof headers === 'object' ? headers : {},
+          size: headers['content-length'],
+          status, // status is http status code
+          time: responseTime,
+        },
+      })
+      .catch((error) => {
+        log.error(
+          `[Symon] Send ${
+            probeState === 'DOWN' ? 'incident' : 'recovery'
+          } event for Alert ID: ${alertId} failed.  ${(error as Error).message}`
+        )
+      })
+  }
 
-    // return unsubscribe function
-    return () => {
-      const index = this.configListeners.indexOf(listener)
-      this.configListeners.splice(index, 1)
+  private async report(): Promise<void> {
+    log.info('[Symon] Report')
+    // Create a task data object
+    const taskData = {
+      apiKey: this.apiKey,
+      hasConnectionToSymon: this.hasConnectionToSymon,
+      httpClient: this.httpClient,
+      monikaId: this.monikaId,
+      probeIds: getProbes().map(({ id }) => id),
+      reportProbesLimit: this.reportProbesLimit,
+      url: this.url,
     }
-  }
 
-  async report(): Promise<void> {
     try {
-      log.debug('Reporting to Symon')
-
-      // Updating requests and notifications to report
-      const probeIds = this.probes.map((probe: Probe) => probe.id)
-
-      // Create a task data object
-      const taskData = {
-        apiKey: this.apiKey,
-        hasConnectionToSymon,
-        httpClient: this.httpClient,
-        monikaId: this.monikaId,
-        probeIds,
-        reportProbesLimit: this.reportProbesLimit,
-        url: this.url,
-      }
-
       // Submit the task to Piscina
       await this.worker.run(JSON.stringify(taskData))
-    } catch (error) {
-      hasConnectionToSymon = false
-      this.configHash = ''
-      log.error("Can't report history to Symon. " + (error as Error).message)
     } finally {
-      setTimeout(async () => {
-        await this.report()
+      this.reportTimeout = setTimeout(() => {
+        this.report
+          .bind(this)()
+          .catch((error) => {
+            this.hasConnectionToSymon = false
+            log.error(`[Symon] Report failed. ${(error as Error).message}`)
+          })
       }, this.reportProbesInterval)
     }
   }
 
   async sendStatus({ isOnline }: { isOnline: boolean }): Promise<void> {
-    try {
-      const response = await this.httpClient({
-        data: {
-          monikaId: this.monikaId,
-          status: isOnline,
-        },
-        method: 'POST',
-        url: '/status',
-      })
-
-      if (response.status === 200) {
-        hasConnectionToSymon = true
-      }
+    const { status } = await this.httpClient({
+      data: {
+        monikaId: this.monikaId,
+        status: isOnline,
+      },
+      method: 'POST',
+      url: '/status',
+    })
 
-      log.debug('Status successfully sent to Symon.')
-    } catch (error: unknown) {
-      log.warn(
-        `Warning: Can't send status to Symon. ${(error as Error).message}`
-      )
+    if (status === 200) {
+      this.hasConnectionToSymon = true
     }
   }
 
-  async stopReport(): Promise<void> {
-    await this.worker.destroy()
-  }
-
   private async fetchProbes() {
-    log.debug('Getting probes from symon')
     const TIMEOUT = 30_000
 
     return this.httpClient
       .get<{ data: Probe[] }>(`/${this.monikaId}/probes`, {
         timeout: TIMEOUT,
         headers: {
-          ...(this.configHash ? { 'If-None-Match': this.configHash } : {}),
+          ...(getContext().config?.version
+            ? { 'If-None-Match': getContext().config?.version }
+            : {}),
         },
         validateStatus(status) {
           return [200, 304].includes(status)
@@ -348,16 +306,15 @@ class SymonClient {
       })
       .then(async (res) => {
         if (!res.data.data) {
-          log.info('No config changes from Symon')
+          log.info('[Symon] No config changes')
 
-          return { probes: this.probes, hash: res.headers.etag }
+          return {
+            probes: getProbes(),
+            hash: res.headers.etag,
+          }
         }
 
-        const validatedProbes = await validateProbes(res.data.data)
-        this.probes = validatedProbes
-        log.info(`Received ${validatedProbes.length} probes`)
-
-        return { probes: validatedProbes, hash: res.headers.etag }
+        return { probes: res.data.data, hash: res.headers.etag }
       })
       .catch((error) => {
         if (error.isAxiosError) {
@@ -375,20 +332,18 @@ class SymonClient {
   }
 
   private async fetchProbesAndUpdateConfig() {
+    log.info('[Symon] Get probes')
     // Fetch the probes
     const { hash, probes } = await this.fetchProbes()
     const newConfig: Config = { probes, version: hash }
-    this.updateConfig(newConfig)
+    await this.setConfig(newConfig)
 
-    // If it has no connection to Symon, set as true
-    // Because it could fetch the probes
-    if (!hasConnectionToSymon) {
-      hasConnectionToSymon = true
-    }
+    // Set connection to symon as true, because it could fetch the probes
+    this.hasConnectionToSymon = true
+    log.info('[Symon] Get probes succeed')
   }
 
   private async handshake(): Promise<string> {
-    log.debug('Performing handshake with symon')
     let handshakeData = await getHandshakeData()
 
     // Check if location id existed and is valid
@@ -414,18 +369,16 @@ class SymonClient {
       .then((res) => res.data?.data.monikaId)
   }
 
-  private updateConfig(newConfig: Config): void {
-    if (newConfig.version && this.configHash !== newConfig.version) {
-      log.debug(`Received config changes. Reloading monika`)
-      this.config = newConfig
-      this.configHash = newConfig.version
-      for (const listener of this.configListeners) {
-        listener(newConfig)
-      }
-    } else {
-      log.debug(`Received config does not change.`)
+  private async setConfig(newConfig: Config) {
+    if (
+      !newConfig.version ||
+      getContext().config?.version === newConfig.version
+    ) {
+      log.info('[Symon] No config change')
+      return
     }
+
+    log.info('[Symon] Config changes. Reloading Monika')
+    await updateConfig(newConfig)
   }
 }
-
-export default SymonClient
diff --git a/src/utils/alert-id.ts b/src/utils/alert-id.ts
index e87263ae..07a3ebb5 100644
--- a/src/utils/alert-id.ts
+++ b/src/utils/alert-id.ts
@@ -22,9 +22,10 @@
  * SOFTWARE.                                                                      *
  **********************************************************************************/
 
-import { getContext } from '../context'
 import type { ValidatedResponse } from '../plugins/validate-response'
 
+import { findProbe } from '../components/config/probe'
+
 export function getAlertID(
   url: string,
   validation: ValidatedResponse,
@@ -34,7 +35,7 @@ export function getAlertID(
     return validation.alert.id
   }
 
-  const probe = getContext().config?.probes.find(({ id }) => id === probeID)
+  const probe = findProbe(probeID)
   if (!probe) {
     return ''
   }


Test patch:
diff --git a/src/components/config/index.test.ts b/src/components/config/index.test.ts
index 8819b14c..81a390c2 100644
--- a/src/components/config/index.test.ts
+++ b/src/components/config/index.test.ts
@@ -68,25 +68,6 @@ describe('getConfig', () => {
     // act and assert
     expect(getConfig()).deep.eq({ probes: [] })
   })
-
-  it('should return config', () => {
-    // arrange
-    const config = {
-      probes: [
-        {
-          id: '1',
-          name: 'example',
-          interval: 1000,
-          alerts: [],
-          requests: [{ body: '', url: 'https://example.com', timeout: 1000 }],
-        },
-      ],
-    }
-    setContext({ config })
-
-    // assert
-    expect(getConfig()).to.deep.eq(config)
-  })
 })
 
 describe('isSymonModeFrom', () => {
diff --git a/src/components/config/probe.test.ts b/src/components/config/probe.test.ts
new file mode 100644
index 00000000..a22bfb6d
--- /dev/null
+++ b/src/components/config/probe.test.ts
@@ -0,0 +1,130 @@
+/**********************************************************************************
+ * MIT License                                                                    *
+ *                                                                                *
+ * Copyright (c) 2021 Hyperjump Technology                                        *
+ *                                                                                *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy   *
+ * of this software and associated documentation files (the "Software"), to deal  *
+ * in the Software without restriction, including without limitation the rights   *
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell      *
+ * copies of the Software, and to permit persons to whom the Software is          *
+ * furnished to do so, subject to the following conditions:                       *
+ *                                                                                *
+ * The above copyright notice and this permission notice shall be included in all *
+ * copies or substantial portions of the Software.                                *
+ *                                                                                *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR     *
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,       *
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE    *
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER         *
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,  *
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE  *
+ * SOFTWARE.                                                                      *
+ **********************************************************************************/
+
+import { expect } from '@oclif/test'
+
+import type { Probe } from '../../interfaces/probe'
+
+import {
+  addProbe,
+  deleteProbe,
+  findProbe,
+  getProbes,
+  setProbes,
+  updateProbe,
+} from './probe'
+
+const probe: Probe = {
+  alerts: [],
+  id: 'xVUcW',
+  interval: 10,
+  name: 'Sample Probe',
+}
+
+describe('Probe cache', () => {
+  beforeEach(() => {
+    for (const { id } of getProbes()) {
+      deleteProbe(id)
+    }
+  })
+  it('should add a probe', () => {
+    // act
+    addProbe(probe)
+
+    // assert
+    expect(findProbe(probe.id)).eq(probe)
+  })
+
+  it('should update a probe', () => {
+    // arrange
+    addProbe(probe)
+    const updatedName = 'Updated Probe'
+
+    // act
+    const isUpdated = updateProbe(probe.id, { ...probe, name: updatedName })
+
+    // assert
+    expect(isUpdated).eq(true)
+    expect(findProbe(probe.id)?.name).eq(updatedName)
+  })
+
+  it('should not update a nonexistent probe', () => {
+    // arrange
+    const updatedName = 'Updated Probe'
+
+    // act
+    const isUpdated = updateProbe('9WpFB', { ...probe, name: updatedName })
+
+    // assert
+    expect(isUpdated).eq(false)
+    expect(findProbe('9WpFB')).undefined
+  })
+
+  it('should get all probes', () => {
+    // arrange
+    addProbe(probe)
+
+    // act
+    const probes = getProbes()
+
+    // assert
+    expect(probes.length).eq(1)
+  })
+
+  it('should not remove a nonexistent probe', () => {
+    // arrange
+    addProbe(probe)
+
+    // act
+    const isDeleted = deleteProbe('9WpFB')
+
+    // assert
+    expect(isDeleted).eq(false)
+    expect(getProbes().length).eq(1)
+  })
+
+  it('should remove a probe', () => {
+    // arrange
+    addProbe(probe)
+
+    // act
+    const isDeleted = deleteProbe(probe.id)
+
+    // assert
+    expect(isDeleted).eq(true)
+    expect(getProbes().length).eq(0)
+  })
+
+  it('should set probes', () => {
+    // arrange
+    addProbe(probe)
+    expect(getProbes().length).eq(1)
+
+    // act
+    setProbes([probe])
+
+    // assert
+    expect(getProbes().length).eq(1)
+  })
+})
diff --git a/src/symon/index.test.ts b/src/symon/index.test.ts
index 4f885436..220467ee 100644
--- a/src/symon/index.test.ts
+++ b/src/symon/index.test.ts
@@ -25,18 +25,38 @@
 import { expect } from '@oclif/test'
 import { rest } from 'msw'
 import { setupServer } from 'msw/node'
-import sinon from 'sinon'
 
-import type { MonikaFlags } from '../flag'
+import { monikaFlagsDefaultValue, type MonikaFlags } from '../flag'
 import type { Config } from '../interfaces/config'
+import type { Probe } from '../interfaces/probe'
 
 import SymonClient from '.'
+import { getContext, resetContext, setContext } from '../context'
+import { deleteProbe, getProbes } from '../components/config/probe'
 import { validateProbes } from '../components/config/validation'
-import * as loggerHistory from '../components/logger/history'
-import { setContext } from '../context'
-
-let getUnreportedLogsStub: sinon.SinonStub
+import events from '../events'
+import { md5Hash } from '../utils/hash'
+import { getEventEmitter } from '../utils/events'
 
+const config: Config = {
+  version: 'asdfg123',
+  probes: [
+    {
+      id: '1',
+      name: 'test',
+      interval: 10,
+      requests: [],
+      alerts: [],
+    },
+    {
+      id: '2',
+      name: 'Example',
+      interval: 10,
+      requests: [{ url: 'https://example.com', body: '', timeout: 2000 }],
+      alerts: [],
+    },
+  ],
+}
 const server = setupServer(
   rest.get('http://ip-api.com/json/192.168.1.1', (_, res, ctx) =>
     res(
@@ -63,6 +83,17 @@ const server = setupServer(
   rest.post('http://localhost:4000/api/v1/monika/status', (_, res, ctx) =>
     res(ctx.status(200))
   ),
+  rest.get('http://localhost:4000/api/v1/monika/1234/probes', (_, res, ctx) =>
+    res(
+      ctx.set('etag', config.version || ''),
+      ctx.json({
+        statusCode: 'ok',
+        message: 'Successfully get probes configuration',
+        data: config.probes,
+      })
+    )
+  ),
+  rest.get('https://example.com', (_, res, ctx) => res(ctx.json({}))),
   rest.get('http://localhost:4000/api/v1/monika/report', (_, res, ctx) =>
     res(
       ctx.json({
@@ -78,38 +109,28 @@ describe('Symon initiate', () => {
     server.listen()
   })
   beforeEach(() => {
+    resetContext()
     setContext({
       flags: {
-        symonUrl: 'https://example.com',
+        symonUrl: 'http://localhost:4000',
         symonKey: 'random-key',
+        symonGetProbesIntervalMs:
+          monikaFlagsDefaultValue.symonGetProbesIntervalMs,
       } as MonikaFlags,
     })
 
-    getUnreportedLogsStub = sinon
-      .stub(loggerHistory, 'getUnreportedLogs')
-      .resolves({ requests: [], notifications: [] })
+    // reset probe cache
+    for (const { id } of getProbes()) {
+      deleteProbe(id)
+    }
   })
   afterEach(() => {
     server.resetHandlers()
-    getUnreportedLogsStub.restore()
   })
   after(() => {
     server.close()
   })
 
-  const config: Config = {
-    version: 'asdfg123',
-    probes: [
-      {
-        id: '1',
-        name: 'test',
-        interval: 10,
-        requests: [],
-        alerts: [],
-      },
-    ],
-  }
-
   it('should send handshake data on initiate', async () => {
     setContext({
       userAgent: 'v1.5.0',
@@ -136,18 +157,6 @@ describe('Symon initiate', () => {
             })
           )
         }
-      ),
-      rest.get(
-        'http://localhost:4000/api/v1/monika/1234/probes',
-        (_, res, ctx) =>
-          res(
-            ctx.set('etag', config.version || ''),
-            ctx.json({
-              statusCode: 'ok',
-              message: 'Successfully get probes configuration',
-              data: config.probes,
-            })
-          )
       )
     )
 
@@ -155,10 +164,8 @@ describe('Symon initiate', () => {
       symonUrl: 'http://localhost:4000',
       symonKey: 'abcd',
     })
-    const reportSpy = sinon.spy(symon, 'report')
     await symon.initiate()
-    await symon.stopReport()
-    expect(symon.monikaId).equals('1234')
+    await symon.stop()
 
     expect(body.publicIp).equals('192.168.1.1')
     expect(body.pid).greaterThan(0)
@@ -170,46 +177,28 @@ describe('Symon initiate', () => {
     expect(body.privateIp).length.greaterThan(0)
     expect(body.os).length.greaterThan(0)
     expect(body.version).equals('v1.5.0')
-
-    expect(reportSpy.called).equals(true)
   }).timeout(15_000)
 
   it('should fetch probes config on initiate', async () => {
-    server.use(
-      rest.get(
-        'http://localhost:4000/api/v1/monika/1234/probes',
-        (_, res, ctx) =>
-          res(
-            ctx.set('etag', config.version || ''),
-            ctx.json({
-              statusCode: 'ok',
-              message: 'Successfully get probes configuration',
-              data: config.probes,
-            })
-          )
-      )
-    )
-
     const symon = new SymonClient({
       symonUrl: 'http://localhost:4000',
       symonKey: 'abcd',
     })
-    sinon.spy(symon, 'report')
 
-    expect(symon.config).to.be.null
+    expect(getContext().config).to.be.undefined
 
     await symon.initiate()
-    await symon.stopReport()
+    await symon.stop()
 
-    expect(symon.config).deep.equals({
+    expect(getContext().config).deep.equals({
       ...config,
       probes: await validateProbes(config.probes),
     })
+    expect(getProbes()).deep.eq(await validateProbes(config.probes))
   }).timeout(15_000)
 
   it('should throw an error if the request to get probes is failed', async () => {
     // arrange
-    server.resetHandlers()
     server.use(
       rest.post(
         'http://localhost:4000/api/v1/monika/client-handshake',
@@ -223,7 +212,10 @@ describe('Symon initiate', () => {
               },
             })
           )
-      )
+      ),
+      rest.get('http://localhost:4000/api/v1/monika/1234/probes', () => {
+        throw new Error('Failed')
+      })
     )
 
     const symon = new SymonClient({
@@ -239,6 +231,8 @@ describe('Symon initiate', () => {
       errorMessage = error?.message
     }
 
+    await symon.stop()
+
     // assert
     expect(errorMessage).eq('Failed to get probes from Symon')
   }).timeout(15_000)
@@ -261,25 +255,229 @@ describe('Symon initiate', () => {
         }
       )
     )
-
     const symon = new SymonClient({
       symonUrl: 'http://localhost:4000',
       symonKey: 'abcd',
+      symonMonikaId: '1234',
     })
-    sinon.spy(symon, 'report')
-    symon.monikaId = '1234'
+    await symon.initiate()
 
     // act
-    await symon.notifyEvent({
-      event: 'incident',
-      alertId: 'alert86',
-      response: { status: 400, time: 1000 },
+    getEventEmitter().emit(events.probe.notification.willSend, {
+      probeState: 'DOWN',
+      validation: {
+        response: {
+          headers: '',
+          status: 400,
+          responseTime: 1000,
+        },
+      },
+      alertId: 'p422i',
     })
+    await symon.stop()
 
     // assert
     expect(body.monikaId).equals('1234')
     expect(body.event).equals('incident')
-    expect(body.alertId).equals('alert86')
-    expect(body.response).deep.equals({ status: 400, time: 1000 })
+    expect(body.alertId).equals('p422i')
+    expect(body.response).deep.equals({
+      headers: {},
+      status: 400,
+      time: 1000,
+    })
+  }).timeout(15_000)
+
+  it('should add a new probe', async () => {
+    // arrange
+    const symonGetProbesIntervalMs = 100
+    setContext({
+      flags: {
+        symonUrl: 'http://localhost:4000',
+        symonKey: 'random-key',
+        symonGetProbesIntervalMs,
+      } as MonikaFlags,
+    })
+    const symon = new SymonClient({
+      symonUrl: 'http://localhost:4000',
+      symonKey: 'abcd',
+    })
+
+    // 1. Check initial probe cache
+    expect(getProbes()).deep.eq([])
+
+    // act
+    // 2. Initiate Symon and get all the probes
+    await symon.initiate()
+
+    // assert
+    // 3. Check the probe data after connected to Symon
+    expect(getProbes()).deep.eq(await validateProbes(config.probes))
+
+    // arrange
+    // 4. Simulate adding a probe
+    const newProbe: Probe = {
+      id: '3',
+      name: 'New Probe',
+      interval: 2,
+      requests: [{ url: 'https://example.com', body: '', timeout: 2000 }],
+      alerts: [],
+    }
+    server.use(
+      rest.get(
+        'http://localhost:4000/api/v1/monika/1234/probes',
+        (_, res, ctx) => {
+          const newProbes: Probe[] = [...config.probes, newProbe]
+
+          return res(
+            ctx.set('etag', md5Hash(newProbes)),
+            ctx.json({
+              statusCode: 'ok',
+              message: 'Successfully get probes configuration',
+              data: newProbes,
+            })
+          )
+        }
+      )
+    )
+
+    // act
+    // 5. Wait for the probe fetch to run
+    await sleep(symonGetProbesIntervalMs)
+
+    // assert
+    // 6. Check the updated probe cache
+    expect(getProbes()).deep.eq(
+      await validateProbes([...config.probes, newProbe])
+    )
+
+    // act
+    // 7. Wait for probe fetch to run
+    await sleep(symonGetProbesIntervalMs)
+
+    // assert
+    // 8. Should not update the probe cache
+    expect(getProbes()).deep.eq(
+      await validateProbes([...config.probes, newProbe])
+    )
+    await symon.stop()
+  }).timeout(15_000)
+
+  it('should update a probe', async () => {
+    // arrange
+    const symonGetProbesIntervalMs = 100
+    setContext({
+      flags: {
+        symonUrl: 'http://localhost:4000',
+        symonKey: 'random-key',
+        symonGetProbesIntervalMs,
+      } as MonikaFlags,
+    })
+    const symon = new SymonClient({
+      symonUrl: 'http://localhost:4000',
+      symonKey: 'abcd',
+    })
+
+    // 1. Check initial probe cache
+    expect(getProbes()).deep.eq([])
+
+    // act
+    // 2. Initiate Symon and get all the probes
+    await symon.initiate()
+
+    // assert
+    // 3. Check the probe data after connected to Symon
+    expect(getProbes()).deep.eq(await validateProbes(config.probes))
+
+    // arrange
+    // 4. Simulate updating a probe
+    const updatedProbes: Probe[] = [{ ...config.probes[0], interval: 5 }]
+    server.use(
+      rest.get(
+        'http://localhost:4000/api/v1/monika/1234/probes',
+        (_, res, ctx) => {
+          const newProbes: Probe[] = updatedProbes
+
+          return res(
+            ctx.set('etag', md5Hash(newProbes)),
+            ctx.json({
+              statusCode: 'ok',
+              message: 'Successfully get probes configuration',
+              data: newProbes,
+            })
+          )
+        }
+      )
+    )
+
+    // act
+    // 5. Wait for the probe fetch to run
+    await sleep(symonGetProbesIntervalMs)
+
+    // assert
+    // 6. Check the updated probe cache
+    expect(getProbes()).deep.eq(await validateProbes(updatedProbes))
+
+    await symon.stop()
+  }).timeout(15_000)
+
+  it('should delete a probe', async () => {
+    // arrange
+    const symonGetProbesIntervalMs = 100
+    setContext({
+      flags: {
+        symonUrl: 'http://localhost:4000',
+        symonKey: 'random-key',
+        symonGetProbesIntervalMs,
+      } as MonikaFlags,
+    })
+    const symon = new SymonClient({
+      symonUrl: 'http://localhost:4000',
+      symonKey: 'abcd',
+    })
+
+    // 1. Check initial probe cache
+    expect(getProbes()).deep.eq([])
+
+    // act
+    // 2. Initiate Symon and get all the probes
+    await symon.initiate()
+
+    // assert
+    // 3. Check the probe data after connected to Symon
+    expect(getProbes()).deep.eq(await validateProbes(config.probes))
+
+    // arrange
+    // 4. Simulate deleting a probe
+    const updatedProbes: Probe[] = config.probes.filter(({ id }) => id === '1')
+    server.use(
+      rest.get(
+        'http://localhost:4000/api/v1/monika/1234/probes',
+        (_, res, ctx) =>
+          res(
+            ctx.set('etag', md5Hash(updatedProbes)),
+            ctx.json({
+              statusCode: 'ok',
+              message: 'Successfully get probes configuration',
+              data: updatedProbes,
+            })
+          )
+      )
+    )
+
+    // act
+    // 5. Wait for the probe fetch to run
+    await sleep(symonGetProbesIntervalMs)
+
+    // assert
+    // 6. Check the updated probe cache
+    expect(getProbes()).deep.eq(await validateProbes(updatedProbes))
+
+    await symon.stop()
   }).timeout(15_000)
 })
+
+function sleep(durationMs: number): Promise<void> {
+  return new Promise((resolve) => {
+    setTimeout(resolve, durationMs)
+  })
+}