Add plugins audit log

src/worker/plugins/run.ts

@@ -7,6 +7,8 @@ export async function runPlugins(server: PluginsServer, event: PluginEvent): Pro
     const pluginsToRun = getPluginsForTeam(server, event.team_id)
     let returnedEvent: PluginEvent | null = event
 
+    const pluginsRan = []
+    const pluginsFailed = []
     for (const pluginConfig of pluginsToRun) {
         const processEvent = await pluginConfig.vm?.getProcessEvent()
 
@@ -15,9 +17,11 @@ export async function runPlugins(server: PluginsServer, event: PluginEvent): Pro
 
             try {
                 returnedEvent = (await processEvent(returnedEvent)) || null
+                pluginsRan.push(pluginConfig.plugin?.name)
             } catch (error) {
                 await processError(server, pluginConfig, error, returnedEvent)
                 server.statsd?.increment(`plugin.${pluginConfig.plugin?.name}.process_event.ERROR`)
+                pluginsFailed.push(pluginConfig.plugin?.name)
             }
             server.statsd?.timing(`plugin.${pluginConfig.plugin?.name}.process_event`, timer)
 
@@ -27,6 +31,11 @@ export async function runPlugins(server: PluginsServer, event: PluginEvent): Pro
         }
     }
 
+    if (!returnedEvent.properties) {
+        returnedEvent.properties = {}
+    }
+    returnedEvent.properties['$plugins_ran'] = pluginsRan
+    returnedEvent.properties['$plugins_failed'] = pluginsFailed
     return returnedEvent
 }
 
@@ -47,16 +56,19 @@ export async function runPluginsOnBatch(server: PluginsServer, batch: PluginEven
         const pluginsToRun = getPluginsForTeam(server, teamId)
 
         let returnedEvents: PluginEvent[] = teamEvents
-
+        const pluginsRan = []
+        const pluginsFailed = []
         for (const pluginConfig of pluginsToRun) {
             const timer = new Date()
             const processEventBatch = await pluginConfig.vm?.getProcessEventBatch()
             if (processEventBatch && returnedEvents.length > 0) {
                 try {
                     returnedEvents = (await processEventBatch(returnedEvents)) || []
+                    pluginsRan.push(pluginConfig.plugin?.name)
                 } catch (error) {
                     await processError(server, pluginConfig, error, returnedEvents[0])
                     server.statsd?.increment(`plugin.${pluginConfig.plugin?.name}.process_event_batch.ERROR`)
+                    pluginsFailed.push(pluginConfig.plugin?.name)
                 }
                 server.statsd?.timing(`plugin.${pluginConfig.plugin?.name}.process_event_batch`, timer)
                 server.statsd?.timing('plugin.process_event_batch', timer, 0.2, {
@@ -66,6 +78,16 @@ export async function runPluginsOnBatch(server: PluginsServer, batch: PluginEven
             }
         }
 
+        for (const event of returnedEvents) {
+            if (event) {
+                if (!event.properties) {
+                    event.properties = {}
+                }
+                event.properties['$plugins_ran'] = pluginsRan
+                event.properties['$plugins_failed'] = pluginsFailed
+            }
+        }
+
         allReturnedEvents = allReturnedEvents.concat(returnedEvents)
     }
 

tests/plugins.test.ts

@@ -115,6 +115,8 @@ test('plugin meta has what it should have', async () => {
     const returnedEvent = await runPlugins(mockServer, event)
 
     expect(Object.keys(returnedEvent!.properties!).sort()).toEqual([
+        '$plugins_failed',
+        '$plugins_ran',
         'attachments',
         'cache',
         'config',
@@ -184,6 +186,72 @@ test('local plugin with broken index.js does not do much', async () => {
     unlink()
 })
 
+test('plugin throwing error does not prevent ingestion and failure is noted in event', async () => {
+    // silence some spam
+    console.log = jest.fn()
+    console.error = jest.fn()
+
+    getPluginRows.mockReturnValueOnce([
+        mockPluginWithArchive(`
+            function processEvent (event) {
+                throw new Error('I always fail!')
+            }
+        `),
+    ])
+    getPluginConfigRows.mockReturnValueOnce([pluginConfig39])
+    getPluginAttachmentRows.mockReturnValueOnce([pluginAttachment1])
+
+    await setupPlugins(mockServer)
+    const { pluginConfigs } = mockServer
+
+    expect(await pluginConfigs.get(39)!.vm!.getTasks()).toEqual({})
+
+    const event = { event: '$test', properties: {}, team_id: 2 } as PluginEvent
+    const returnedEvent = await runPlugins(mockServer, { ...event })
+
+    const expectedReturnEvent = {
+        ...event,
+        properties: {
+            $plugins_ran: [],
+            $plugins_failed: ['test-maxmind-plugin'],
+        },
+    }
+    expect(returnedEvent).toEqual(expectedReturnEvent)
+})
+
+test('events have property $plugins_ran set to the plugins that succeeded', async () => {
+    // silence some spam
+    console.log = jest.fn()
+    console.error = jest.fn()
+
+    getPluginRows.mockReturnValueOnce([
+        mockPluginWithArchive(`
+            function processEvent (event) {
+                return event
+            }
+        `),
+    ])
+    getPluginConfigRows.mockReturnValueOnce([pluginConfig39])
+    getPluginAttachmentRows.mockReturnValueOnce([pluginAttachment1])
+
+    await setupPlugins(mockServer)
+    const { pluginConfigs } = mockServer
+
+    expect(await pluginConfigs.get(39)!.vm!.getTasks()).toEqual({})
+
+    const event = { event: '$test', properties: {}, team_id: 2 } as PluginEvent
+    const returnedEvent = await runPlugins(mockServer, { ...event })
+
+    const expectedReturnEvent = {
+        ...event,
+        properties: {
+            $plugins_ran: ['test-maxmind-plugin'],
+            $plugins_failed: [],
+        },
+    }
+    expect(returnedEvent).toEqual(expectedReturnEvent)
+})
+
 test('archive plugin with broken plugin.json does not do much', async () => {
     // silence some spam
     console.log = jest.fn()