feat(data-warehouse): log entries by schema (#23686)

Co-authored-by: github-actions <41898282+github-actions[bot]@users.noreply.github.com>

frontend/src/lib/api.ts

@@ -2044,6 +2044,16 @@ const api = {
         async incremental_fields(schemaId: ExternalDataSourceSchema['id']): Promise<SchemaIncrementalFieldsResponse> {
             return await new ApiRequest().externalDataSourceSchema(schemaId).withAction('incremental_fields').create()
         },
+        async logs(
+            schemaId: ExternalDataSourceSchema['id'],
+            params: LogEntryRequestParams = {}
+        ): Promise<PaginatedResponse<LogEntry>> {
+            return await new ApiRequest()
+                .externalDataSourceSchema(schemaId)
+                .withAction('logs')
+                .withQueryString(params)
+                .get()
+        },
     },
 
     dataWarehouseViewLinks: {

frontend/src/scenes/data-warehouse/settings/source/DataWarehouseSourceSettingsScene.tsx

@@ -62,7 +62,7 @@ export function DataWarehouseSourceSettingsScene(): JSX.Element {
                 activeKey={currentTab}
                 onChange={(tab) => setCurrentTab(tab as DataWarehouseSourceSettingsTabs)}
                 tabs={Object.entries(TabContent).map(([tab, ContentComponent]) => ({
-                    label: FriendlyTabNames[tab],
+                    label: FriendlyTabNames[tab as DataWarehouseSourceSettingsTabs],
                     key: tab,
                     content: <ContentComponent />,
                 }))}

frontend/src/scenes/data-warehouse/settings/source/Logs.tsx

@@ -0,0 +1,86 @@
+import { LemonButton, LemonTable, LemonTableColumns } from '@posthog/lemon-ui'
+import { useActions, useValues } from 'kea'
+import { LOGS_PORTION_LIMIT } from 'lib/constants'
+import { dayjs } from 'lib/dayjs'
+import { pluralize } from 'lib/utils'
+import { LogLevelDisplay } from 'scenes/pipeline/utils'
+
+import { ExternalDataJob, LogEntry } from '~/types'
+
+import { schemaLogLogic } from './schemaLogLogic'
+
+const columns: LemonTableColumns<LogEntry> = [
+    {
+        title: 'Timestamp',
+        key: 'timestamp',
+        dataIndex: 'timestamp',
+        width: 1,
+        render: (_, entry) => dayjs(entry.timestamp).format('YYYY-MM-DD HH:mm:ss.SSS UTC'),
+    },
+    {
+        title: 'Level',
+        key: 'level',
+        dataIndex: 'level',
+        width: 1,
+        render: (_, entry) => LogLevelDisplay(entry.level),
+    },
+    {
+        title: 'Run ID',
+        key: 'run_id',
+        dataIndex: 'instance_id',
+        width: 1,
+        render: (_, entry) => entry.instance_id,
+    },
+    {
+        title: 'Message',
+        key: 'message',
+        dataIndex: 'message',
+        width: 6,
+    },
+]
+
+interface LogsTableProps {
+    job: ExternalDataJob
+}
+
+export const LogsView = ({ job }: LogsTableProps): JSX.Element => {
+    const logic = schemaLogLogic({ job })
+    const { logs, logsLoading, logsBackground, isThereMoreToLoad } = useValues(logic)
+    const { revealBackground, loadSchemaLogsMore } = useActions(logic)
+
+    return (
+        <div className="ph-no-capture space-y-2 flex-1">
+            <LemonButton
+                onClick={revealBackground}
+                loading={logsLoading}
+                type="secondary"
+                fullWidth
+                center
+                disabledReason={!logsBackground.length ? "There's nothing to load" : undefined}
+            >
+                {logsBackground.length
+                    ? `Load ${pluralize(logsBackground.length, 'newer entry', 'newer entries')}`
+                    : 'No new entries'}
+            </LemonButton>
+            <LemonTable
+                dataSource={logs}
+                columns={columns}
+                loading={logsLoading}
+                className="ph-no-capture"
+                pagination={{ pageSize: 200, hideOnSinglePage: true }}
+            />
+            {!!logs.length && (
+                <LemonButton
+                    onClick={loadSchemaLogsMore}
+                    loading={logsLoading}
+                    type="secondary"
+                    fullWidth
+                    center
+                    disabledReason={!isThereMoreToLoad ? "There's nothing more to load" : undefined}
+                >
+                    {isThereMoreToLoad ? `Load up to ${LOGS_PORTION_LIMIT} older entries` : 'No older entries'}
+                </LemonButton>
+            )}
+        </div>
+    )
+}

frontend/src/scenes/data-warehouse/settings/source/Syncs.tsx

@@ -5,6 +5,7 @@ import { useValues } from 'kea'
 import { ExternalDataJob } from '~/types'
 
 import { dataWarehouseSourceSettingsLogic } from './dataWarehouseSourceSettingsLogic'
+import { LogsView } from './Logs'
 
 const StatusTagSetting: Record<ExternalDataJob['status'], LemonTagType> = {
     Running: 'primary',
@@ -46,6 +47,19 @@ export const Syncs = (): JSX.Element => {
                     },
                 },
             ]}
+            expandable={
+                jobs.length > 0
+                    ? {
+                          expandedRowRender: (job) => (
+                              <div className="p-4">
+                                  <LogsView job={job} />
+                              </div>
+                          ),
+                          rowExpandable: () => true,
+                          noIndent: true,
+                      }
+                    : undefined
+            }
         />
     )
 }

frontend/src/scenes/data-warehouse/settings/source/schemaLogLogic.ts

@@ -0,0 +1,144 @@
+import { actions, events, kea, key, listeners, path, props, reducers, selectors } from 'kea'
+import { loaders } from 'kea-loaders'
+import api from 'lib/api'
+import { LOGS_PORTION_LIMIT } from 'lib/constants'
+
+import { ExternalDataJob, ExternalDataSourceSchema, LogEntry, LogEntryLevel } from '~/types'
+
+import type { schemaLogLogicType } from './schemaLogLogicType'
+
+export const ALL_LOG_LEVELS: LogEntryLevel[] = ['DEBUG', 'LOG', 'INFO', 'WARNING', 'ERROR']
+export const DEFAULT_LOG_LEVELS: LogEntryLevel[] = ['DEBUG', 'LOG', 'INFO', 'WARNING', 'ERROR']
+
+export interface SchemaLogLogicProps {
+    job: ExternalDataJob
+}
+
+export const schemaLogLogic = kea<schemaLogLogicType>([
+    path(['scenes', 'data-warehouse', 'settings', 'source', 'schemaLogLogic']),
+    props({} as SchemaLogLogicProps),
+    key(({ job }) => job.id),
+    actions({
+        clearBackgroundLogs: true,
+        setLogLevelFilters: (levelFilters: LogEntryLevel[]) => ({ levelFilters }),
+        setSearchTerm: (searchTerm: string) => ({ searchTerm }),
+        setSchema: (schemaId: ExternalDataSourceSchema['id']) => ({ schemaId }),
+        markLogsEnd: true,
+    }),
+    loaders(({ values, actions, cache, props }) => ({
+        logs: {
+            __default: [] as LogEntry[],
+            loadSchemaLogs: async () => {
+                const response = await api.externalDataSchemas.logs(props.job.schema.id, {
+                    level: values.levelFilters.join(','),
+                    search: values.searchTerm,
+                    instance_id: props.job.id,
+                })
+
+                if (!cache.pollingInterval) {
+                    cache.pollingInterval = setInterval(actions.loadSchemaLogsBackgroundPoll, 2000)
+                }
+
+                return response.results
+            },
+            loadSchemaLogsMore: async () => {
+                if (!values.selectedSchemaId) {
+                    return []
+                }
+                const response = await api.externalDataSchemas.logs(values.selectedSchemaId, {
+                    level: values.levelFilters.join(','),
+                    search: values.searchTerm,
+                    instance_id: props.job.id,
+                    before: values.leadingEntry?.timestamp,
+                })
+
+                if (response.results.length < LOGS_PORTION_LIMIT) {
+                    actions.markLogsEnd()
+                }
+
+                return [...values.logs, ...response.results]
+            },
+            revealBackground: () => {
+                const newArray = [...values.logsBackground, ...values.logs]
+                actions.clearBackgroundLogs()
+                return newArray
+            },
+        },
+        logsBackground: {
+            __default: [] as LogEntry[],
+            loadSchemaLogsBackgroundPoll: async () => {
+                const response = await api.externalDataSchemas.logs(props.job.schema.id, {
+                    level: values.levelFilters.join(','),
+                    search: values.searchTerm,
+                    instance_id: props.job.id,
+                    after: values.leadingEntry?.timestamp,
+                })
+
+                return [...response.results, ...values.logsBackground]
+            },
+        },
+    })),
+    reducers({
+        levelFilters: [
+            DEFAULT_LOG_LEVELS,
+            {
+                setLogLevelFilters: (_, { levelFilters }) => levelFilters,
+            },
+        ],
+        searchTerm: [
+            '',
+            {
+                setSearchTerm: (_, { searchTerm }) => searchTerm,
+            },
+        ],
+        selectedSchemaId: [
+            null as string | null,
+            {
+                setSchema: (_, { schemaId }) => schemaId,
+            },
+        ],
+        isThereMoreToLoad: [
+            true,
+            {
+                loadSchemaLogsSuccess: (_, { logs }) => logs.length >= LOGS_PORTION_LIMIT,
+                markLogsEnd: () => false,
+            },
+        ],
+    }),
+    selectors({
+        leadingEntry: [
+            (s) => [s.logs, s.logsBackground],
+            (logs, logsBackground): LogEntry | null => {
+                if (logsBackground.length) {
+                    return logsBackground[0]
+                }
+                if (logs.length) {
+                    return logs[0]
+                }
+                return null
+            },
+        ],
+    }),
+    listeners(({ actions }) => ({
+        setLogLevelFilters: () => {
+            actions.loadSchemaLogs()
+        },
+        setSearchTerm: async ({ searchTerm }, breakpoint) => {
+            if (searchTerm) {
+                await breakpoint(1000)
+            }
+            actions.loadSchemaLogs()
+        },
+        setSchema: () => {
+            actions.loadSchemaLogs()
+        },
+    })),
+    events(({ actions, cache }) => ({
+        afterMount: () => {
+            actions.loadSchemaLogs()
+        },
+        beforeUnmount: () => {
+            clearInterval(cache.pollingInterval)
+        },
+    })),
+])