Adds Pipeline error/warning log indicator to list page

[eblairmckee]

Added pipeline log indicators to the pipelines list page. Note: the data layer implementation is a temporary solution until we have an Oxla integration that provides proper database queries for logs filtered by pipelineId, log level, etc. and/or we get additional backend functionality from Cloud.

Pipeline log indicators and related react-query calls are gated behind the enableNewPipelineLogs feature flag.

Why This Is Temporary

Right now, we just have the Cloud API messages backend (same API we use for the Topics messages viewer). We're using this because all Connect pipelines on a cluster produce logs to the __redpanda.connect.logs where the key is the pipelineId and the value is the log. To get any meaningful value out of pipeline logs we need to do a lot of filtering, however the only filtering that exists on this API is a custom javascript function you can execute server-side. The rest has to happen at run-time.

Ideally, we want to store logs in a database, so we can use Oxla to make complex queries like:

SELECT * FROM logs WHERE pipeline_id = ? AND level IN ('WARN', 'ERROR')

Current Implementation

Architecture

┌─────────────────────────────────────────────────────────────────────┐
│ __redpanda.connect.logs (Kafka Topic)                               │
│ All pipeline logs written here with pipeline_id as message key      │
└───────────────────────────────┬─────────────────────────────────────┘
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────────┐
│ useStreamingPipelineLogCounts Hook                                  │
│ - Streams from topic with server side JS filter                     │
│ - Fetches last N hours of logs                                      │
│ - maxResults = pipelineIds.length × 100                             │
└───────────────────────────────┬─────────────────────────────────────┘
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────────┐
│ aggregateMessages                                                   │
│ - Filters for WARN/ERROR levels (client-side)                       │
│ - Categorizes by scope: input | output | root                       │
│ - Groups counts by pipeline_id                                      │
└───────────────────────────────┬─────────────────────────────────────┘
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────────┐
│ PipelineLogIndicator Component                                      │
│ - Displays error/warning badges with counts                         │
│ - Shown in Name, Input, and Output columns                          │
└─────────────────────────────────────────────────────────────────────┘

Log Scope Categorization

Logs are categorized by their path field:

Path Pattern	Scope	Column
root.input.*	input	Input
root.output.*	output	Output
Everything else	root	Pipeline Name

This allows users to see at a glance whether issues are occurring in a pipeline's input connector, output connector, or elsewhere.

Files Changed

File	Purpose
list.tsx	Main list component, enriches pipelines with log counts
pipeline-messages.tsx	useStreamingPipelineLogCounts hook
messages.tsx	useListMessagesStream hook for Kafka streaming
pipeline-log-indicator.tsx	Badge component with error/warning counts
logs/constants.ts	LOG_PATH_INPUT, LOG_PATH_OUTPUT constants

The "Noisy Pipeline" Problem

Critical Caveat: All pipeline logs are written to a single Kafka topic (__redpanda.connect.logs). When streaming logs:

1. We request maxResults = pipelineCount × 100 messages (max on the backend is 500, but that's currently being increased by Julin)
2. Messages are returned in order (newest first or by partition)
3. A single pipeline generating many logs can dominate the sample

Example: If Pipeline A generates 1000 logs/minute and Pipeline B generates 10 logs/minute:

• With maxResults = 200, Pipeline A's logs may entirely fill the sample
• Pipeline B's error/warning indicators may show as 0 even if issues exist

Alternative Approaches Considered

1. Increase Sample Size (Current Direction)

Approach: Increase maxResults multiplier from 100 to 500+ per pipeline.

Pros	Cons
Simple change	More data transferred
Better representation	Still susceptible to noisy pipelines
	Memory pressure on client

2. Parallel Queries Per Pipeline

Approach: Run one query per pipeline, fetching 100 most recent logs each.

// Pseudocode
const results = await Promise.all(
  pipelineIds.map((id) => fetchLogs({ pipelineId: id, maxResults: 100 }))
);

Pros	Cons
Each pipeline gets fair representation	N queries instead of 1
No noisy pipeline problem	Not performant for 20+ pipelines
	Server load concerns

3. Time-Windowed Sampling

Approach: Fetch logs from multiple time windows and merge.

Pros	Cons
Better distribution over time	Increased complexity
	Multiple queries

4. Server-Side Aggregation (Ideal)

Approach: Backend endpoint that returns aggregated counts.

GET / api / pipelines / log - counts;
// Returns: { pipelineId: { errors: N, warnings: N } }

Pros	Cons
Minimal data transfer	Requires backend work
Accurate counts	Still limited without Oxla
Scalable

Smart Polling for Status Updates

The list view implements smart polling to keep pipeline states current:

useListPipelinesQuery(undefined, {
  enableSmartPolling: true,
});

How It Works

1. After data fetch, check if any pipeline is in a transitional state (STARTING or STOPPING)
2. If yes: poll every 2 seconds (SHORT_POLLING_INTERVAL)
3. If no: disable polling (use React Query's default cache behavior)

This ensures users see timely state transitions without unnecessary polling when all pipelines are stable.

Transitional States

// pipeline/constants.ts
export const TRANSITIONAL_STATES = [
  Pipeline_State.STARTING,
  Pipeline_State.STOPPING,
] as const;

The actions dropdown also shows "Retry start" / "Retry stop" options for pipelines stuck in transitional states.

[eblairmckee]

have an action item for myself to upstream this (as well as a few other components that are missing ! bangs for border colors)

[eblairmckee]

I've been noodling on this for awhile, and even when we used a FilterIcon here, I would still forget that these were filters. Changes to a chevron/dropdown icon, and I really liked the way it looks in both the new Pipelines list table, as well as other tables that use it. Eventually I want to come up with a better pattern for filtered tables, but that is getting punted until phase 1 of this project is done.

[malinskibeniamin]

To me the most intuitive way is https://ui.bazza.dev/docs/data-table-filter

[eblairmckee]

eventually I want to upstream this into the ui-registry (Dot and a docs example of Badge with Dot)

[eblairmckee]

also want to upstream this into design system as Status or StatusBadge

[r-vasquez]

Should it be State? or the filter be Status? I see in the screenshot that the filter label is State but the column name is Status

[malinskibeniamin]

Let's be careful with this. This needs QA effort to ensure it works because streaming with rpc requires special handling