From 257a804412e76870e27a878259e1732360219bbd Mon Sep 17 00:00:00 2001 From: ritikbansod2912 Date: Sun, 27 Sep 2026 02:45:46 +0530 Subject: [PATCH] feat(search): whole-topic background message search (UI, REST, CLI, MCP) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements F2 from the deep analysis — the biggest explorer gap: finding a specific message deep in a topic. A bounded background job scans every partition (or one) forward from a timestamp or the beginning, collecting matches — read-only like the browser (throwaway group, never commits). - POST /api/clusters/{id}/topics/{topic}/search starts a job (regex / value-contains / key-contains filters, optional from/to timestamp bounds, limit); GET .../searches/{id} polls progress + results; DELETE cancels. Bounded: max 1000 results, 2M scanned messages, 3 concurrent jobs, jobs expire from memory after 30 minutes. - Matches carry the same schema-decoded view as the browser (shared MessageBrowserService.toView), so results feed the existing message detail modal and reproduce flow. - Web UI: new Search tab in the data explorer with progress polling. - CLI: kview search [...] and kview search-status . - MCP: kview_search_messages (start + wait for completion) and kview_search_status — 18 tools now. - Role model: search POSTs are read-shaped, so a read-only role may search; the mutation audit trail intentionally does not log searches (they read, they write nothing). - tests: 4 integration tests against an embedded broker (regex across 3 partitions, early stop at limit, timestamp bounds, unknown id). Full suite 53/53. --- CHANGELOG.md | 1 + README.md | 5 +- cli/kview.mjs | 42 +++ mcp/kview-mcp.js | 57 ++++ .../kafkawrapper/auth/MutationRequests.java | 3 +- .../browse/MessageBrowserService.java | 5 +- .../kafkawrapper/search/SearchController.java | 44 +++ .../kafkawrapper/search/SearchService.java | 279 ++++++++++++++++++ .../resources/static/js/views/explorer.js | 105 ++++++- .../search/SearchIntegrationTest.java | 112 +++++++ 10 files changed, 645 insertions(+), 8 deletions(-) create mode 100644 src/main/java/com/ritikbansod/kafkawrapper/search/SearchController.java create mode 100644 src/main/java/com/ritikbansod/kafkawrapper/search/SearchService.java create mode 100644 src/test/java/com/ritikbansod/kafkawrapper/search/SearchIntegrationTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 6993dad..71b233d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- **Whole-topic message search** (F2): `POST /api/clusters/{clusterId}/topics/{topic}/search` starts a bounded background scan (regex / contains / key / timestamp filters, capped at 1000 results and 2M scanned messages, max 3 concurrent jobs) with progress polling and cancellation (`GET`/`DELETE .../searches/{id}`). Exposed in the UI as a new **Search** tab in the explorer, in the CLI as `kview search` / `kview search-status`, and in MCP as `kview_search_messages` + `kview_search_status` (18 tools now). Read-only like the browser: throwaway group, never commits. - **API authentication with roles** (`kview.auth.*`, off by default): `token` mode (static bearer tokens) and `oidc` mode (JWTs from any issuer, configurable role claim), each with an **admin** and a **read-only** role — reads for everyone, mutations (produce, topic/group management, offset resets, connection management) admin-only. The web UI prompts for the token (🔑 in the top bar), the CLI takes `--token`/`KVIEW_TOKEN`, the MCP server takes `KVIEW_TOKEN`. In authenticated modes CORS is closed unless `KVIEW_AUTH_ALLOWED_ORIGINS` lists origins; the static UI and `/actuator/health` stay public. See the new Authentication section in the README. - **AI/MCP safety pack**: global read-only mode (`KVIEW_READONLY=true` — every mutation refused for all clients and roles, UI shows a banner), an append-only mutation audit trail (`KVIEW_AUDIT=true` → `/audit.log` JSONL with client, method, path and status; all three clients send an `X-Kview-Client` identifier), a public `GET /api/meta` endpoint, and a per-tool allowlist for the MCP server (`KVIEW_ALLOWED_TOOLS`, e.g. serve only read-only tools to an AI agent). diff --git a/README.md b/README.md index 5f2afa1..53eb731 100644 --- a/README.md +++ b/README.md @@ -42,7 +42,8 @@ consumer group, so it is safe to point at production topics. - **Topics**: leaders, replicas, ISR, begin/end offsets, message counts and configs. Create and delete topics, increase partitions, edit configs. - **Data explorer**: read messages (latest N, exact offsets, from a timestamp, filter by - key or value contains), pretty-printed JSON with a copy button, live tail over SSE, + key or value contains), whole-topic background search with regex filters and progress, + pretty-printed JSON with a copy button, live tail over SSE, reproduce a message to any topic with one click. - **Produce**: key, value, partition, timestamp and headers, with JSON validation as you type. - **Consumer groups**: state, per-partition lag with bars, member assignments, delete a @@ -116,7 +117,7 @@ scripting. If the server runs with authentication enabled, add `--token ` (en ### 3. MCP server (AI assistants) `mcp/kview-mcp.js` is a [Model Context Protocol](https://modelcontextprotocol.io) server -that exposes Kview as 16 tools, so Claude Desktop, Cursor, ZCode or any MCP client can +that exposes Kview as 18 tools, so Claude Desktop, Cursor, ZCode or any MCP client can inspect and operate your Kafka cluster in natural language: cluster health, topic inventory, message browsing with filters, producing test events, consumer-group lag and broker distribution. Zero dependencies, speaks newline-delimited JSON-RPC over stdio. diff --git a/cli/kview.mjs b/cli/kview.mjs index 868a8bf..b2195ce 100644 --- a/cli/kview.mjs +++ b/cli/kview.mjs @@ -64,6 +64,17 @@ function fail(msg) { process.exit(1); } +/** Starts are async: poll the search until done or the wait budget runs out. */ +async function pollSearch(searchId, waitSec) { + const deadline = Date.now() + waitSec * 1000; + let status; + do { + status = await api('GET', clusterPath(`/searches/${encodeURIComponent(searchId)}`)); + if (status.state === 'RUNNING') await new Promise((r) => setTimeout(r, 500)); + } while (status.state === 'RUNNING' && Date.now() < deadline); + return status; +} + // ---- output helpers ---------------------------------------------------------- const oneLine = (s, max = 70) => s === null || s === undefined ? '' : @@ -111,6 +122,9 @@ Topics: topics | topic | topic-create [--partitions N] [--rf N] Messages: produce [-k key] [-v payload] [--file path] [--partition N] [--header k=v...] browse [--start latest|earliest|timestamp] [--limit N] [--partition N] [--key-contains S] [--value-contains S] [--ts 2026-01-01T10:00:00Z] + search [--value-regex RE] [--key-contains S] [--value-contains S] + [--from-ts MS|ISO] [--to-ts MS|ISO] [--limit N] [--partition N] [--wait SEC] + search-status tail [--from latest|earliest] [--partition N] (Ctrl+C stops) Groups: groups | lag | reset-offsets --topic T --mode earliest|latest|offset|timestamp [--value V] | group-delete @@ -227,6 +241,34 @@ Examples: kview overview (out.reachedEnd ? ' (end of partition reached)' : ' (older messages may exist — raise --limit)')); }, + async search() { + const topic = positional[1] ?? fail('usage: kview search [--value-regex RE] [--key-contains S] [--value-contains S] [--from-ts MS|ISO] [--to-ts MS|ISO] [--limit N] [--partition N] [--wait SEC]'); + const toMs = (v) => (v === undefined ? undefined : (/^-?\d{13,}$/.test(v) ? Number(v) : Date.parse(v))); + const body = { + keyContains: flags['key-contains'], + valueContains: flags['value-contains'], + valueRegex: flags['value-regex'], + fromTs: toMs(flags['from-ts']), + toTs: toMs(flags['to-ts']), + limit: num('limit'), + partition: num('partition'), + }; + const { searchId } = await api('POST', clusterPath(`/topics/${encodeURIComponent(topic)}/search`), body); + const status = await pollSearch(searchId, num('wait') ?? 60); + if (JSON_OUT) { console.log(JSON.stringify(status, null, 2)); return; } + for (const m of status.results ?? []) console.log(msgLine(m) + decodedSuffix(m)); + console.error(`-- ${status.matched} matched · scanned ${status.scanned} · state ${status.state}` + + (status.state === 'RUNNING' ? ` — still running, poll with: kview search-status ${searchId}` : '') + + (status.limitReached ? ' · limit reached' : '')); + }, + + async 'search-status'() { + const searchId = positional[1] ?? fail('usage: kview search-status '); + const status = await api('GET', clusterPath(`/searches/${encodeURIComponent(searchId)}`)); + console.log(JSON_OUT ? JSON.stringify(status, null, 2) + : `search ${status.searchId} state=${status.state} matched=${status.matched} scanned=${status.scanned} partitions=${status.partitionsDone}/${status.partitionsTotal}`); + }, + async tail() { const topic = positional[1] ?? fail('usage: kview tail [--from latest|earliest] [--partition N]'); const qs = new URLSearchParams({ from: flags.from ?? 'latest' }); diff --git a/mcp/kview-mcp.js b/mcp/kview-mcp.js index 7661bd8..9e66f0f 100644 --- a/mcp/kview-mcp.js +++ b/mcp/kview-mcp.js @@ -174,6 +174,37 @@ const impl = { return api(`/api/clusters/${cluster(p)}/groups`); }, + async search(p) { + const topic = requireArg(p, 'topic'); + const body = {}; + for (const k of ['keyContains', 'valueContains', 'valueRegex']) { + if (p[k] !== undefined) body[k] = String(p[k]); + } + for (const k of ['fromTs', 'toTs']) { + if (p[k] !== undefined) body[k] = Number(p[k]); + } + if (p.limit !== undefined) body.limit = Math.min(Number(p.limit) || 100, 1000); + if (p.partition !== undefined) body.partition = Number(p.partition); + const { searchId } = await api(`/api/clusters/${cluster(p)}/topics/${encodeURIComponent(topic)}/search`, + { method: 'POST', body: JSON.stringify(body) }); + const maxWaitMs = Math.min(Number(p.maxWaitMs) || 15000, 60000); + const deadline = Date.now() + maxWaitMs; + let status; + do { + await new Promise((r) => setTimeout(r, 400)); + status = await api(`/api/clusters/${cluster(p)}/searches/${searchId}`); + } while (status.state === 'RUNNING' && Date.now() < deadline); + if (status.state === 'RUNNING') { + status.note = `Still running (scanned ${status.scanned} so far) — poll with kview_search_status(searchId='${searchId}')`; + } + return status; + }, + + async searchStatus(p) { + const searchId = requireArg(p, 'searchId'); + return api(`/api/clusters/${cluster(p)}/searches/${searchId}`); + }, + async groupDetail(p) { const group = requireArg(p, 'group'); return api(`/api/clusters/${cluster(p)}/groups/${encodeURIComponent(group)}`); @@ -310,6 +341,32 @@ const TOOLS = [ }, ['topic']), run: impl.browse, }, + { + name: 'kview_search_messages', + description: 'Search a whole topic (all partitions, background job) for messages matching text or regex filters, optionally bounded by timestamps. Waits up to maxWaitMs for completion and returns matches with progress. Read-only.', + inputSchema: object({ + topic: str('Topic name'), + valueRegex: str('Regular expression the value must match, e.g. "orderId"\\s*:\\s*"ORD-[0-9]+" (optional)'), + valueContains: str('Only messages whose value contains this text (optional)'), + keyContains: str('Only messages whose key contains this text (optional)'), + fromTs: num('Epoch-ms lower timestamp bound (inclusive, optional)'), + toTs: num('Epoch-ms upper timestamp bound (exclusive, optional)'), + limit: num('Max matches to collect (default 100, max 1000)'), + partition: num('Restrict to one partition (optional)'), + maxWaitMs: num('How long to wait for completion, ms (default 15000, max 60000)'), + cluster: str('Kview connection id (optional)'), + }, ['topic']), + run: impl.search, + }, + { + name: 'kview_search_status', + description: 'Progress and results of a background search started with kview_search_messages (searchId is in its output).', + inputSchema: object({ + searchId: str('Search id from kview_search_messages'), + cluster: str('Kview connection id (optional)'), + }, ['searchId']), + run: impl.searchStatus, + }, { name: 'kview_consumer_groups', description: 'List consumer groups with state, committed partition count and total lag.', diff --git a/src/main/java/com/ritikbansod/kafkawrapper/auth/MutationRequests.java b/src/main/java/com/ritikbansod/kafkawrapper/auth/MutationRequests.java index a54385b..4f3aa49 100644 --- a/src/main/java/com/ritikbansod/kafkawrapper/auth/MutationRequests.java +++ b/src/main/java/com/ritikbansod/kafkawrapper/auth/MutationRequests.java @@ -18,7 +18,8 @@ final class MutationRequests implements RequestMatcher { static final MutationRequests INSTANCE = new MutationRequests(); private static final PathMatcher ANT = new AntPathMatcher(); - private static final List READ_SHAPED_POSTS = List.of("/**/browse", "/**/decode", "/**/encode", "/**/test"); + private static final List READ_SHAPED_POSTS = + List.of("/**/browse", "/**/decode", "/**/encode", "/**/test", "/**/search"); private MutationRequests() { } diff --git a/src/main/java/com/ritikbansod/kafkawrapper/browse/MessageBrowserService.java b/src/main/java/com/ritikbansod/kafkawrapper/browse/MessageBrowserService.java index 659950f..47f342b 100644 --- a/src/main/java/com/ritikbansod/kafkawrapper/browse/MessageBrowserService.java +++ b/src/main/java/com/ritikbansod/kafkawrapper/browse/MessageBrowserService.java @@ -158,7 +158,8 @@ private List resolvePartitions(KafkaConsumer con return result; } - private BrowserMessage toView(String clusterId, ConsumerRecord record) { + /** Maps a raw record to the API view (schema decode included) — shared with the background search. */ + public BrowserMessage toView(String clusterId, ConsumerRecord record) { byte[] keyBytes = record.key(); byte[] valueBytes = record.value(); Map headers = new LinkedHashMap<>(); @@ -178,7 +179,7 @@ private BrowserMessage toView(String clusterId, ConsumerRecord r schema); } - static String asText(byte[] bytes) { + public static String asText(byte[] bytes) { return bytes == null ? null : new String(bytes, StandardCharsets.UTF_8); } diff --git a/src/main/java/com/ritikbansod/kafkawrapper/search/SearchController.java b/src/main/java/com/ritikbansod/kafkawrapper/search/SearchController.java new file mode 100644 index 0000000..030d359 --- /dev/null +++ b/src/main/java/com/ritikbansod/kafkawrapper/search/SearchController.java @@ -0,0 +1,44 @@ +package com.ritikbansod.kafkawrapper.search; + +import org.springframework.web.bind.annotation.DeleteMapping; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +import java.util.Map; + +/** + * Whole-topic background search: start (POST — a read-shaped mutation), poll + * progress + results, cancel. The search itself is read-only: never commits, + * never joins a real group. + */ +@RestController +@RequestMapping("/api/clusters/{clusterId}") +public class SearchController { + + private final SearchService searchService; + + public SearchController(SearchService searchService) { + this.searchService = searchService; + } + + @PostMapping("/topics/{topic}/search") + public Map start(@PathVariable String clusterId, @PathVariable String topic, + @RequestBody SearchService.SearchRequest request) { + return Map.of("searchId", searchService.start(clusterId, topic, request)); + } + + @GetMapping("/searches/{searchId}") + public SearchService.SearchStatus status(@PathVariable String clusterId, @PathVariable String searchId) { + return searchService.status(clusterId, searchId); + } + + @DeleteMapping("/searches/{searchId}") + public Map cancel(@PathVariable String clusterId, @PathVariable String searchId) { + searchService.cancel(clusterId, searchId); + return Map.of("searchId", searchId, "status", "cancelling"); + } +} diff --git a/src/main/java/com/ritikbansod/kafkawrapper/search/SearchService.java b/src/main/java/com/ritikbansod/kafkawrapper/search/SearchService.java new file mode 100644 index 0000000..2c938b4 --- /dev/null +++ b/src/main/java/com/ritikbansod/kafkawrapper/search/SearchService.java @@ -0,0 +1,279 @@ +package com.ritikbansod.kafkawrapper.search; + +import com.ritikbansod.kafkawrapper.browse.MessageBrowserService; +import com.ritikbansod.kafkawrapper.connection.KafkaClusterManager; +import jakarta.annotation.PreDestroy; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.consumer.OffsetAndTimestamp; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.WakeupException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Service; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.NoSuchElementException; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.regex.Pattern; + +/** + * Whole-topic background search: assigns every partition (or one), seeks to the + * optional from-timestamp, and scans forward up to a snapshot end offset, + * collecting matches — read-only like the browser (own throwaway group, never + * commits). Bounded: results capped, scanned messages capped, a small pool of + * concurrent jobs, jobs expire from memory. + */ +@Service +public class SearchService { + + private static final Logger log = LoggerFactory.getLogger(SearchService.class); + private static final int MAX_CONCURRENT_JOBS = 3; + private static final int MAX_RESULTS = 1000; + private static final long MAX_SCANNED = 2_000_000; + private static final long JOB_TTL_MS = 30 * 60_000L; + + private final KafkaClusterManager manager; + private final MessageBrowserService browser; + private final ExecutorService pool = Executors.newFixedThreadPool(2, r -> { + Thread thread = new Thread(r, "kview-search"); + thread.setDaemon(true); + return thread; + }); + private final Map jobs = new ConcurrentHashMap<>(); + + public SearchService(KafkaClusterManager manager, MessageBrowserService browser) { + this.manager = manager; + this.browser = browser; + } + + public record SearchRequest(Long fromTs, Long toTs, String keyContains, String valueContains, + String valueRegex, Integer partition, Integer limit) { } + + public record SearchStatus(String searchId, String topic, String state, String error, + int partitionsTotal, int partitionsDone, long scanned, long matched, + boolean limitReached, long startedAt, long finishedAt, + List results) { } + + /** Starts a search and returns its id. */ + public synchronized String start(String clusterId, String topic, SearchRequest request) { + long running = jobs.values().stream().filter(j -> "RUNNING".equals(j.state)).count(); + if (running >= MAX_CONCURRENT_JOBS) { + throw new IllegalArgumentException( + "Too many concurrent searches (" + MAX_CONCURRENT_JOBS + ") — wait or cancel one first"); + } + if (request.valueRegex() != null && !request.valueRegex().isBlank()) { + Pattern.compile(request.valueRegex()); // invalid regex fails fast with a 400 + } + String id = UUID.randomUUID().toString().substring(0, 8); + SearchJob job = new SearchJob(id, clusterId, topic, request); + job.limit = request.limit() == null ? 100 : Math.min(Math.max(request.limit(), 1), MAX_RESULTS); + jobs.put(id, job); + job.future = pool.submit(() -> run(job)); + log.info("Search {} started on cluster '{}' topic '{}'", id, clusterId, topic); + return id; + } + + public SearchStatus status(String clusterId, String searchId) { + SearchJob job = job(clusterId, searchId); + List snapshot; + synchronized (job.results) { + snapshot = List.copyOf(job.results); + } + return new SearchStatus(job.id, job.topic, job.state, job.error, + job.partitionsTotal.get(), job.partitionsDone.get(), + job.scanned.get(), job.matched.get(), job.matched.get() >= job.limit, + job.startedAt, job.finishedAt, snapshot); + } + + public void cancel(String clusterId, String searchId) { + SearchJob job = job(clusterId, searchId); + job.cancelled.set(true); + KafkaConsumer consumer = job.consumer; + if (consumer != null) consumer.wakeup(); + } + + /** Drops finished jobs from memory after their TTL. */ + @Scheduled(fixedDelay = 60_000) + void cleanup() { + long now = System.currentTimeMillis(); + jobs.entrySet().removeIf(e -> !"RUNNING".equals(e.getValue().state) + && now - e.getValue().startedAt > JOB_TTL_MS); + } + + private SearchJob job(String clusterId, String searchId) { + SearchJob job = jobs.get(searchId); + if (job == null || !job.clusterId.equals(clusterId)) { + throw new NoSuchElementException("Unknown search id '" + searchId + "'"); + } + return job; + } + + private void run(SearchJob job) { + SearchRequest request = job.request; + String clusterId = job.clusterId; + int limit = job.limit; + Pattern regex = request.valueRegex() == null || request.valueRegex().isBlank() + ? null : Pattern.compile(request.valueRegex()); + String keyContains = lower(request.keyContains()); + String valueContains = lower(request.valueContains()); + + Map props; + try { + props = new HashMap<>(manager.get(clusterId).consumerProperties()); + } catch (Exception e) { + job.finish("FAILED", e.getMessage()); + return; + } + // same read-only contract as the browser: throwaway group, never commits + props.put(ConsumerConfig.GROUP_ID_CONFIG, "kview-search-" + job.id); + props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500); + + try (KafkaConsumer consumer = new KafkaConsumer<>(props)) { + job.consumer = consumer; + List partitions = partitions(consumer, job.topic, request.partition()); + if (partitions.isEmpty()) { + throw new org.apache.kafka.common.errors.UnknownTopicOrPartitionException(job.topic); + } + job.partitionsTotal.set(partitions.size()); + + Map ends = consumer.endOffsets(partitions); + Map beginnings = consumer.beginningOffsets(partitions); + consumer.assign(partitions); + for (TopicPartition tp : partitions) { + if (request.fromTs() != null) { + OffsetAndTimestamp at = consumer.offsetsForTimes( + Map.of(tp, request.fromTs())).get(tp); + consumer.seek(tp, at == null ? ends.get(tp) : Math.max(at.offset(), beginnings.getOrDefault(tp, 0L))); + } else { + consumer.seek(tp, beginnings.getOrDefault(tp, 0L)); + } + } + + while (!job.cancelled.get() + && job.scanned.get() < MAX_SCANNED + && job.matched.get() < limit + && job.partitionsDone.get() < partitions.size()) { + for (ConsumerRecord record : consumer.poll(Duration.ofMillis(250))) { + TopicPartition tp = new TopicPartition(record.topic(), record.partition()); + if (job.donePartitions.containsKey(tp)) continue; + if (request.toTs() != null && record.timestamp() > request.toTs()) { + markDone(job, tp); + continue; + } + job.scanned.incrementAndGet(); + if (matches(record, keyContains, valueContains, regex)) { + synchronized (job.results) { + job.results.add(browser.toView(clusterId, record)); + } + job.matched.incrementAndGet(); + if (job.matched.get() >= limit) break; + } + } + // a partition is done when the scan reached its snapshot end offset + for (TopicPartition tp : partitions) { + if (!job.donePartitions.containsKey(tp) + && consumer.position(tp) >= ends.getOrDefault(tp, 0L)) { + markDone(job, tp); + } + } + } + job.finish(job.cancelled.get() ? "CANCELLED" : "DONE", null); + } catch (WakeupException e) { + job.finish("CANCELLED", null); + } catch (Exception e) { + log.warn("Search {} failed: {}", job.id, e.getMessage()); + job.finish("FAILED", e.getMessage()); + } + } + + private void markDone(SearchJob job, TopicPartition tp) { + if (job.donePartitions.putIfAbsent(tp, true) == null) { + job.partitionsDone.incrementAndGet(); + } + } + + private boolean matches(ConsumerRecord record, String keyContains, String valueContains, + Pattern regex) { + String key = MessageBrowserService.asText(record.key()); + String value = MessageBrowserService.asText(record.value()); + if (keyContains != null && (key == null || !key.toLowerCase().contains(keyContains))) return false; + if (valueContains != null && (value == null || !value.toLowerCase().contains(valueContains))) return false; + return regex == null || (value != null && regex.matcher(value).find()); + } + + private List partitions(KafkaConsumer consumer, String topic, Integer only) { + List result = new ArrayList<>(); + var infos = consumer.partitionsFor(topic, Duration.ofSeconds(10)); + if (infos != null) { + infos.forEach(info -> { + if (only == null || only == info.partition()) { + result.add(new TopicPartition(topic, info.partition())); + } + }); + } + return result; + } + + private static String lower(String s) { + return s == null || s.isBlank() ? null : s.toLowerCase(); + } + + @PreDestroy + void shutdown() { + pool.shutdownNow(); + } + + private static final class SearchJob { + final String id; + final String clusterId; + final String topic; + final SearchRequest request; + int limit = 100; + final long startedAt = System.currentTimeMillis(); + final AtomicLong scanned = new AtomicLong(); + final AtomicLong matched = new AtomicLong(); + final AtomicInteger partitionsTotal = new AtomicInteger(); + final AtomicInteger partitionsDone = new AtomicInteger(); + final AtomicBoolean cancelled = new AtomicBoolean(); + final Map donePartitions = new ConcurrentHashMap<>(); + final List results = Collections.synchronizedList(new ArrayList<>()); + volatile String state = "RUNNING"; + volatile String error; + volatile long finishedAt; + volatile KafkaConsumer consumer; + volatile Future future; + + SearchJob(String id, String clusterId, String topic, SearchRequest request) { + this.id = id; + this.clusterId = clusterId; + this.topic = topic; + this.request = request; + } + + void finish(String state, String error) { + if ("RUNNING".equals(this.state)) { + this.state = state; + this.error = error; + this.finishedAt = System.currentTimeMillis(); + } + } + } +} diff --git a/src/main/resources/static/js/views/explorer.js b/src/main/resources/static/js/views/explorer.js index 45ad28f..9cf51a2 100644 --- a/src/main/resources/static/js/views/explorer.js +++ b/src/main/resources/static/js/views/explorer.js @@ -1,5 +1,5 @@ -// ===== Data Explorer: browse (seek-based), live tail (SSE), produce ===== -import { get, post, clusterPath, loadTopics, getToken } from '../api.js'; +// ===== Data Explorer: browse (seek-based), search (whole topic), live tail (SSE), produce ===== +import { get, post, put, del, clusterPath, loadTopics, getToken } from '../api.js'; import { esc, fmtNum, fmtTs, fmtRel, badge, spinner, toast, modal, jsonBlock, previewValue, copyText, focusable, debounce, } from '../ui.js'; @@ -10,6 +10,8 @@ let tailTimer = null; export async function renderExplorer(view, query = {}) { window.__tailClose?.(); window.__tailClose = null; + window.__searchClose?.(); + window.__searchClose = null; if (tailTimer) { clearInterval(tailTimer); tailTimer = null; } view.innerHTML = spinner(); const tab = query.tab || 'browse'; @@ -29,6 +31,7 @@ export async function renderExplorer(view, query = {}) {
+
@@ -45,7 +48,7 @@ export async function renderExplorer(view, query = {}) { (currentTab !== 'browse' ? `&tab=${currentTab}` : '')); const show = (which, autoRun = false) => - ({ browse: renderBrowse, tail: renderTail, produce: renderProduce })[which](panel, topicSelect, () => partitionCount, autoRun); + ({ browse: renderBrowse, search: renderSearch, tail: renderTail, produce: renderProduce })[which](panel, topicSelect, () => partitionCount, autoRun); view.querySelector('#ex-tabs').addEventListener('click', (e) => { const btn = e.target.closest('[data-tab]'); @@ -275,6 +278,102 @@ function messageDetailModal(m) { }); } +// ---------------- Search (whole topic, background job) ---------------- + +function renderSearch(panel, topicSelect, getPartitions) { + window.__searchClose?.(); + panel.innerHTML = ` +
+
+
+
+
+
+
+
+
+
+
+ + +
+
+
Scans every partition of the topic in the background — capped at 1000 results / 2M messages scanned. Read-only, never commits.
+
+
Run a search to see matches
`; + + const statusEl = panel.querySelector('#se-status'); + const resultsEl = panel.querySelector('#se-results'); + const goBtn = panel.querySelector('#se-go'); + const stopBtn = panel.querySelector('#se-stop'); + let timer = null; + + const stop = () => { + if (timer) { clearInterval(timer); timer = null; } + stopBtn.disabled = true; + goBtn.disabled = false; + }; + window.__searchClose = () => { if (timer) { clearInterval(timer); timer = null; } }; + + stopBtn.addEventListener('click', async () => { + const id = stopBtn.dataset.searchId; + if (id) await del(clusterPath() + `/searches/${id}`).catch(() => { }); + stop(); + }); + + const paint = (results) => { + resultsEl.innerHTML = results.length === 0 + ? '
No matches yet…
' + : `
+ + ${results.map((m, i) => ` + + + + + + + + `).join('')} +
PartitionOffsetTimeKeyValue
${m.partition}${m.offset}${fmtRel(m.timestamp)}${esc(m.key ?? 'null')}${previewValue(m.value)}${codecChip(m)}
`; + resultsEl.querySelectorAll('tr[data-msg]').forEach((tr) => { + tr.addEventListener('click', () => messageDetailModal(results[Number(tr.dataset.msg)])); + }); + }; + + goBtn.addEventListener('click', async () => { + const topic = topicSelect.value; + const dt = (id) => { const v = panel.querySelector(id).value; return v ? new Date(v).getTime() : null; }; + const body = { + partition: panel.querySelector('#se-partition').value === '' ? null : Number(panel.querySelector('#se-partition').value), + valueRegex: panel.querySelector('#se-regex').value.trim() || null, + valueContains: panel.querySelector('#se-value').value.trim() || null, + keyContains: panel.querySelector('#se-key').value.trim() || null, + fromTs: dt('#se-from'), + toTs: dt('#se-to'), + limit: Number(panel.querySelector('#se-limit').value) || 100, + }; + try { + const { searchId } = await post(clusterPath() + `/topics/${encodeURIComponent(topic)}/search`, body); + stopBtn.dataset.searchId = searchId; + stopBtn.disabled = false; + goBtn.disabled = true; + timer = setInterval(async () => { + try { + const s = await get(clusterPath() + `/searches/${searchId}`); + statusEl.innerHTML = `${esc(s.state)} · partitions ${s.partitionsDone}/${s.partitionsTotal}` + + ` · scanned ${fmtNum(s.scanned)} · matched ${s.matched}${s.limitReached ? ' · limit reached' : ''}` + + (s.error ? ` · ${esc(s.error)}` : ''); + paint(s.results || []); + if (s.state !== 'RUNNING') stop(); + } catch (e) { statusEl.textContent = e.message; stop(); } + }, 600); + } catch (err) { + statusEl.textContent = err.message; + } + }); +} + // ---------------- Live tail ---------------- function renderTail(panel, topicSelect, getPartitions) { diff --git a/src/test/java/com/ritikbansod/kafkawrapper/search/SearchIntegrationTest.java b/src/test/java/com/ritikbansod/kafkawrapper/search/SearchIntegrationTest.java new file mode 100644 index 0000000..02b2dac --- /dev/null +++ b/src/test/java/com/ritikbansod/kafkawrapper/search/SearchIntegrationTest.java @@ -0,0 +1,112 @@ +package com.ritikbansod.kafkawrapper.search; + +import com.ritikbansod.kafkawrapper.connection.KafkaClusterManager; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.test.context.DynamicPropertyRegistry; +import org.springframework.test.context.DynamicPropertySource; + +import java.util.NoSuchElementException; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Whole-topic background search against an embedded broker: regex + contains + * filters across partitions, limit, timestamp bounds and unknown ids. + * One topic per test — searches see the full topic, so they must not share data. + */ +@SpringBootTest(properties = "kview.data-dir=target/search-test-data") +@EmbeddedKafka(partitions = 3, topics = {"search-regex", "search-limit", "search-ts"}) +class SearchIntegrationTest { + + @Autowired + SearchService searchService; + + @Autowired + KafkaClusterManager clusterManager; + + @DynamicPropertySource + static void kafkaBootstrap(DynamicPropertyRegistry registry) { + registry.add("spring.kafka.bootstrap-servers", + () -> System.getProperty("spring.embedded.kafka.brokers")); + } + + @Test + void regexSearchFindsMatchesAcrossAllPartitions() throws Exception { + produce("search-regex", System.currentTimeMillis() - 30_000, 90); + + String id = searchService.start(KafkaClusterManager.DEFAULT_CLUSTER_ID, "search-regex", + new SearchService.SearchRequest(null, null, null, null, "\"orderId\"\\s*:\\s*\"ORD-7[0-9]", null, null)); + SearchService.SearchStatus status = awaitDone(id); + + assertThat(status.state()).isEqualTo("DONE"); + // ORD-70..ORD-79 → 10 matches, spread over 3 partitions + assertThat(status.matched()).isEqualTo(10); + assertThat(status.partitionsTotal()).isEqualTo(3); + assertThat(status.partitionsDone()).isEqualTo(3); + assertThat(status.scanned()).isEqualTo(90); + assertThat(status.results()).allSatisfy(m -> assertThat(m.value()).contains("ORD-7")); + assertThat(status.results().get(0).partition()).isBetween(0, 2); + } + + @Test + void limitStopsTheScanEarly() throws Exception { + produce("search-limit", System.currentTimeMillis() - 30_000, 90); + + String id = searchService.start(KafkaClusterManager.DEFAULT_CLUSTER_ID, "search-limit", + new SearchService.SearchRequest(null, null, null, "amount", null, null, 5)); + SearchService.SearchStatus status = awaitDone(id); + + assertThat(status.state()).isEqualTo("DONE"); + assertThat(status.matched()).isEqualTo(5); + assertThat(status.limitReached()).isTrue(); + assertThat(status.scanned()).isLessThan(90); + } + + @Test + void timestampBoundsNarrowTheScan() throws Exception { + long base = System.currentTimeMillis() - 30_000; + produce("search-ts", base, 90); + + String id = searchService.start(KafkaClusterManager.DEFAULT_CLUSTER_ID, "search-ts", + new SearchService.SearchRequest(System.currentTimeMillis() + 60_000, null, null, null, null, null, null)); + SearchService.SearchStatus nothingAtOrAfter = awaitDone(id); + assertThat(nothingAtOrAfter.state()).isEqualTo("DONE"); + assertThat(nothingAtOrAfter.matched()).isZero(); // every record is older than the fromTs bound + + String id2 = searchService.start(KafkaClusterManager.DEFAULT_CLUSTER_ID, "search-ts", + new SearchService.SearchRequest(base - 1000, System.currentTimeMillis(), null, "amount", null, null, null)); + SearchService.SearchStatus all = awaitDone(id2); + assertThat(all.matched()).isEqualTo(90); + } + + @Test + void unknownSearchIdIsReported() { + assertThatThrownBy(() -> searchService.status(KafkaClusterManager.DEFAULT_CLUSTER_ID, "nope")) + .isInstanceOf(NoSuchElementException.class); + } + + private void produce(String topic, long baseTs, int count) throws Exception { + var template = clusterManager.get(KafkaClusterManager.DEFAULT_CLUSTER_ID).template(); + for (int i = 0; i < count; i++) { + String payload = "{\"orderId\":\"ORD-" + i + "\",\"amount\":" + i + ".0}"; + template.send(new ProducerRecord<>(topic, i % 3, baseTs + i, ("k-" + i).getBytes(), + payload.getBytes())).get(); + } + } + + private SearchService.SearchStatus awaitDone(String id) throws Exception { + long deadline = System.currentTimeMillis() + 30_000; + SearchService.SearchStatus status = searchService.status(KafkaClusterManager.DEFAULT_CLUSTER_ID, id); + while ("RUNNING".equals(status.state()) && System.currentTimeMillis() < deadline) { + Thread.sleep(100); + status = searchService.status(KafkaClusterManager.DEFAULT_CLUSTER_ID, id); + } + return status; + } +}