Use query endpoint (#1792)

* refactor search to use query

Signed-off-by: Mason Fish <mason@brimsecurity.com>

* use header for format instead of qParam

Signed-off-by: Mason Fish <mason@brimsecurity.com>

* rework types, fix lint

Signed-off-by: Mason Fish <mason@brimsecurity.com>

* adjust test setup and stubs

Signed-off-by: Mason Fish <mason@brimsecurity.com>

* generate query responses

Signed-off-by: Mason Fish <mason@brimsecurity.com>

* adjust response generation

* fix unit tests, update snapshots

Signed-off-by: Mason Fish <mason@brimsecurity.com>

* fix api tests

Signed-off-by: Mason Fish <mason@brimsecurity.com>

* fix abort handling

Signed-off-by: Mason Fish <mason@brimsecurity.com>

Co-authored-by: Mason Fish <mason@brimsecurity.com>

app/search/flows/__snapshots__/histogram-search.test.ts.snap

@@ -5,9 +5,9 @@ Object {
   "keys": Array [
     "capture_loss",
     "conn",
+    "files",
     "x509",
     "ssl",
-    "files",
     "dns",
     "weird",
     "stats",

app/search/flows/histogram-search.test.ts

@@ -30,16 +30,18 @@ beforeEach(() => {
     Pools.setDetail("1", pool)
   ])
   store.dispatch(tabHistory.push(lakePath(pool.id, "1")))
-  zealot.stubStream("search", countByPathResp)
+  zealot.stubStream("query", countByPathResp)
 })
 
 const submit = () => dispatch(histogramSearch())
 
 test("zealot gets the request", async () => {
   await submit()
-  const calls = zealot.calls("search")
+  const calls = zealot.calls("query")
   expect(calls.length).toBe(1)
-  expect(calls[0].args).toEqual("* | every 12h count() by _path")
+  expect(calls[0].args).toEqual(
+    "from '1' | ts >= 2015-03-05T14:15:00Z | ts <= 2015-04-13T09:36:33.751Z | * | every 12h count() by _path"
+  )
 })
 
 test("the chart status updates", async () => {

app/search/flows/md5-search.ts

@@ -19,12 +19,12 @@ export const md5Search = (md5: string): Thunk<BrimSearch> => (
   const poolId = Current.getPoolId(getState())
   if (!poolId) return
   const [from, to] = Tab.getSpanAsDates(getState())
-  const query = parallelizeProcs([
+  const q = parallelizeProcs([
     filenameCorrelation(md5),
     md5Correlation(md5),
     rxHostsCorrelation(md5),
     txHostsCorrelation(md5)
   ])
 
-  return dispatch(search({id, query, from, to, poolId}))
+  return dispatch(search({id, query: q, from, to, poolId}))
 }

app/search/flows/next-page-viewer-search.test.ts

@@ -23,7 +23,7 @@ const records = [
 let store, zealot, tabId
 beforeEach(() => {
   zealot = createZealotMock()
-  zealot.stubStream("search", [])
+  zealot.stubStream("query", [])
   store = initTestStore(zealot.zealot)
   tabId = Tabs.getActive(store.getState())
   const ws = fixtures("workspace1")
@@ -48,30 +48,30 @@ test("#fetchNextPage dispatches splice", () => {
 })
 
 test("#fetchNextPage adds 1ms to ts of last change", () => {
-  const search = jest.spyOn(zealot.zealot, "search")
+  const search = jest.spyOn(zealot.zealot, "query")
   store.dispatch(nextPageViewerSearch())
 
   const data = records[1].at(1) as zed.Time
   const lastChangeTs = data.toDate().getTime()
   expect(search).toHaveBeenCalledWith(
-    expect.any(String),
+    expect.stringContaining(
+      `ts <= ${new Date(lastChangeTs + 1).toISOString()}`
+    ),
     expect.objectContaining({
-      from: new Date(0),
-      to: new Date(lastChangeTs + 1)
+      signal: expect.any(Object)
     })
   )
 })
 
 test("#fetchNextPage when there is only 1 event", () => {
-  const search = jest.spyOn(zealot.zealot, "search")
+  const search = jest.spyOn(zealot.zealot, "query")
   store.dispatch(Viewer.splice(tabId, 1))
   store.dispatch(nextPageViewerSearch())
 
   expect(search).toHaveBeenCalledWith(
     expect.any(String),
     expect.objectContaining({
-      from: new Date(0),
-      to: new Date(10 * 1000)
+      signal: expect.any(Object)
     })
   )
 })

app/search/flows/viewer-search.test.ts

@@ -6,18 +6,19 @@ import Pools from "src/js/state/Pools"
 import Viewer from "src/js/state/Viewer"
 import Workspaces from "src/js/state/Workspaces"
 import fixtures from "test/unit/fixtures"
-import initTestStore from "test/unit/helpers/initTestStore"
-import responses from "test/unit/responses"
+import initTestStore, {TestStore} from "test/unit/helpers/initTestStore"
 import {createRecord} from "test/shared/factories/zed-factory"
 import {useResponse} from "test/shared/responses"
-import {createZealotMock} from "zealot"
+import {createZealotMock, ZealotMock} from "zealot"
 import {viewerSearch} from "./viewer-search"
 
 const dnsResp = useResponse("dns")
 const pool = fixtures("pool1")
-const warningResp = responses("search_warning.txt")
+const warningResp = useResponse("searchWarning")
 
-let store, zealot, dispatch, select
+let dispatch, select
+let zealot: ZealotMock
+let store: TestStore
 beforeEach(() => {
   zealot = createZealotMock()
   store = initTestStore(zealot.zealot)
@@ -37,27 +38,30 @@ beforeEach(() => {
   store.dispatch(tabHistory.push(lakePath(pool.id, "1")))
 })
 
+const from = new Date()
+const to = new Date(1)
 const submit = () =>
   dispatch(
     viewerSearch({
       query: "dns query | head 500",
-      from: new Date(),
-      to: new Date(1)
+      from,
+      to
     })
   )
 
 describe("a normal response", () => {
   beforeEach(() => {
-    zealot.stubStream("search", dnsResp)
+    zealot.stubStream("query", dnsResp)
   })
 
   test("zealot gets the request", async () => {
     await submit()
-    const calls = zealot.calls("search")
+    const calls = zealot.calls("query")
     expect(calls.length).toBe(1)
-    expect(calls[0].args).toEqual("dns query | head 500")
+    expect(calls[0].args).toEqual(
+      `from '1' | ts >= ${from.toISOString()} | ts <= 1970-01-01T00:00:00.001Z | dns query | head 500`
+    )
   })
-
   test("the table gets populated", async () => {
     await submit()
     expect(select(Viewer.getViewerRecords).length).toBe(2)
@@ -92,9 +96,9 @@ describe("a normal response", () => {
 })
 
 test("a response with a warning", async () => {
-  zealot.stubStream("search", warningResp)
+  zealot.stubStream("query", warningResp)
   await submit()
   expect(select(SearchBar.getSearchBarError)).toBe(
-    "Cut field boo not present in input"
+    "Sort field boo not present in input"
   )
 })

app/search/flows/viewer-search.ts

@@ -1,6 +1,5 @@
 import {ANALYTIC_MAX_RESULTS, PER_PAGE} from "src/js/flows/config"
 import {search} from "src/js/flows/search/mod"
-import {SearchResponse} from "src/js/flows/search/response"
 import ErrorFactory from "src/js/models/ErrorFactory"
 import Columns from "src/js/state/Columns"
 import Current from "src/js/state/Current"
@@ -11,6 +10,7 @@ import {Thunk} from "src/js/state/types"
 import Viewer from "src/js/state/Viewer"
 import {SchemaMap} from "src/js/state/Viewer/types"
 import {zed} from "zealot"
+import {SearchResponse} from "../../../src/js/flows/search/response"
 
 type Args = {
   query: string
@@ -70,7 +70,7 @@ function handle(
         dispatch(Viewer.updateColumns(tabId, schemas))
         dispatch(Columns.touch(schemas))
       })
-      .warnings((warning) => dispatch(SearchBar.errorSearchBarParse(warning)))
+      .warning((warning) => dispatch(SearchBar.errorSearchBarParse(warning)))
       .error((error) => {
         dispatch(Notice.set(ErrorFactory.create(error)))
       })

ppl/detail/flows/fetch.test.ts

@@ -27,27 +27,27 @@ describe("zeek log when community_id is found", () => {
   let setup
   beforeEach(async () => {
     setup = await loginTo("workspace1", "pool1")
-    setup.zealot.stubStream("search", stubs.uidResult)
-    setup.zealot.stubStream("search", stubs.uidAndCommunityResult)
+    setup.zealot.stubStream("query", stubs.uidResult)
+    setup.zealot.stubStream("query", stubs.uidAndCommunityResult)
   })
 
   test("runs two queries ", async () => {
     const {zealot, store} = setup
-    const before = zealot.calls("search").length
+    const before = zealot.calls("query").length
     await store.dispatch(fetchCorrelation(zeek))
-    expect(zealot.calls("search")).toHaveLength(before + 2)
+    expect(zealot.calls("query")).toHaveLength(before + 2)
   })
 
   test("executes uid first, then cid", async () => {
     const {store, zealot} = setup
     await store.dispatch(fetchCorrelation(zeek))
-    const searches = zealot.calls("search")
+    const searches = zealot.calls("query")
     const len = searches.length
     expect(searches[len - 2].args).toMatchInlineSnapshot(
-      `"uid==\\"CbOjYpkXn9LfqV51c\\" or \\"CbOjYpkXn9LfqV51c\\" in conn_uids or \\"CbOjYpkXn9LfqV51c\\" in uids or referenced_file.uid==\\"CbOjYpkXn9LfqV51c\\" | head 100"`
+      `"from '1' | ts >= 2015-03-05T14:15:00Z | ts <= 2015-04-13T09:36:33.751Z | uid==\\"CbOjYpkXn9LfqV51c\\" or \\"CbOjYpkXn9LfqV51c\\" in conn_uids or \\"CbOjYpkXn9LfqV51c\\" in uids or referenced_file.uid==\\"CbOjYpkXn9LfqV51c\\" | head 100"`
     )
     expect(searches[len - 1].args).toMatchInlineSnapshot(
-      `"uid==\\"CbOjYpkXn9LfqV51c\\" or \\"CbOjYpkXn9LfqV51c\\" in conn_uids or \\"CbOjYpkXn9LfqV51c\\" in uids or referenced_file.uid==\\"CbOjYpkXn9LfqV51c\\" or (community_id == \\"1:N7YGmWjwTmMKNhsZHBR618n3ReA=\\" and ts >= 1582646593.978 and ts < 1582646683.994) | head 100"`
+      `"from '1' | ts >= 2015-03-05T14:15:00Z | ts <= 2015-04-13T09:36:33.751Z | uid==\\"CbOjYpkXn9LfqV51c\\" or \\"CbOjYpkXn9LfqV51c\\" in conn_uids or \\"CbOjYpkXn9LfqV51c\\" in uids or referenced_file.uid==\\"CbOjYpkXn9LfqV51c\\" or (community_id == \\"1:N7YGmWjwTmMKNhsZHBR618n3ReA=\\" and ts >= 1582646593.978 and ts < 1582646683.994) | head 100"`
     )
   })
 
@@ -62,21 +62,21 @@ describe("zeek log when community_id is not found", () => {
   let setup
   beforeEach(async () => {
     setup = await loginTo("workspace1", "pool1")
-    setup.zealot.stubStream("search", stubs.noCommunityIdInConn)
+    setup.zealot.stubStream("query", stubs.noCommunityIdInConn)
   })
 
   test("executes only 1 query", async () => {
     const {zealot, store} = setup
-    const before = zealot.calls("search").length
+    const before = zealot.calls("query").length
     await store.dispatch(fetchCorrelation(zeek))
-    expect(zealot.calls("search")).toHaveLength(before + 1)
+    expect(zealot.calls("query")).toHaveLength(before + 1)
   })
 
   test("runs the uid search", async () => {
     const {zealot, store} = setup
     await store.dispatch(fetchCorrelation(zeek))
-    expect(last<any>(zealot.calls("search")).args).toMatchInlineSnapshot(
-      `"uid==\\"CbOjYpkXn9LfqV51c\\" or \\"CbOjYpkXn9LfqV51c\\" in conn_uids or \\"CbOjYpkXn9LfqV51c\\" in uids or referenced_file.uid==\\"CbOjYpkXn9LfqV51c\\" | head 100"`
+    expect(last<any>(zealot.calls("query")).args).toMatchInlineSnapshot(
+      `"from '1' | ts >= 2015-03-05T14:15:00Z | ts <= 2015-04-13T09:36:33.751Z | uid==\\"CbOjYpkXn9LfqV51c\\" or \\"CbOjYpkXn9LfqV51c\\" in conn_uids or \\"CbOjYpkXn9LfqV51c\\" in uids or referenced_file.uid==\\"CbOjYpkXn9LfqV51c\\" | head 100"`
     )
   })
 
@@ -92,21 +92,21 @@ describe("suricata alert when community_id found", () => {
 
   beforeEach(async () => {
     setup = await loginTo("workspace1", "pool1")
-    setup.zealot.stubStream("search", stubs.alertResults)
+    setup.zealot.stubStream("query", stubs.alertResults)
   })
 
   test("issues only one search", async () => {
     const {zealot, store} = setup
-    const before = zealot.calls("search").length
+    const before = zealot.calls("query").length
     await store.dispatch(fetchCorrelation(suricata))
-    expect(zealot.calls("search")).toHaveLength(before + 1)
+    expect(zealot.calls("query")).toHaveLength(before + 1)
   })
 
   test("issues the community id query", async () => {
     const {zealot, store} = setup
     await store.dispatch(fetchCorrelation(suricata))
-    expect(last<any>(zealot.calls("search")).args).toMatchInlineSnapshot(
-      `"community_id==\\"1:N7YGmWjwTmMKNhsZHBR618n3ReA=\\" | head 100"`
+    expect(last<any>(zealot.calls("query")).args).toMatchInlineSnapshot(
+      `"from '1' | ts >= 2015-03-05T14:15:00Z | ts <= 2015-04-13T09:36:33.751Z | community_id==\\"1:N7YGmWjwTmMKNhsZHBR618n3ReA=\\" | head 100"`
     )
   })
 

scripts/test/search.js

@@ -19,9 +19,10 @@ withLake(
   async (zealot) => {
     try {
       const poolId = await ingest(zealot)
-      const from = new Date(0)
-      const to = new Date()
-      const search = await zealot.search(QUERY, {poolId, from, to})
+      const from = new Date(0).toISOString()
+      const to = new Date().toISOString()
+      const annotatedQuery = `from '${poolId}' | ts >= ${from} | ts <= ${to} | ${QUERY}`
+      const search = await zealot.query(annotatedQuery)
       const text = await search.origResp.text()
       console.log(text)
     } catch (e) {

src/js/components/ExportModal.tsx

@@ -10,7 +10,7 @@ import ToolbarButton from "../../../app/toolbar/button"
 import exportResults from "../flows/exportResults"
 import {ipcRenderer} from "electron"
 import {useDispatch} from "react-redux"
-import {SearchFormat} from "../../../zealot"
+import {QueryFormat} from "../../../zealot"
 import InputLabel from "./common/forms/InputLabel"
 import {defaultModalButton} from "../../../test/integration/helpers/locators"
 import {toast} from "react-hot-toast"
@@ -69,7 +69,7 @@ const ExportModal = ({onClose}) => {
     if (canceled) return
 
     toast.promise(
-      dispatch(exportResults(filePath, format as SearchFormat)),
+      dispatch(exportResults(filePath, format as QueryFormat)),
       {
         loading: "Exporting...",
         success: "Export Complete",

src/js/flows/exportResults.ts

@@ -1,6 +1,6 @@
 import fs from "fs"
 import http from "http"
-import {SearchFormat} from "zealot"
+import {QueryFormat} from "zealot"
 import brim from "../brim"
 import Columns from "../state/Columns"
 import Current from "../state/Current"
@@ -9,6 +9,7 @@ import SystemTest from "../state/SystemTest"
 import Tab from "../state/Tab"
 import {Thunk} from "../state/types"
 import {getZealot} from "./getZealot"
+import {annotateQuery} from "./search/mod"
 
 function cutColumns(program, columns) {
   if (columns.allVisible()) {
@@ -30,7 +31,7 @@ export function prepareProgram(format, program, columns) {
 
 export default (
   filePath: string,
-  format: SearchFormat
+  format: QueryFormat
 ): Thunk<Promise<string>> => (dispatch, getState) => {
   const zealot = dispatch(getZealot())
   const poolId = Current.getPoolId(getState())
@@ -44,13 +45,17 @@ export default (
   const {host, port} = Current.mustGetWorkspace(getState())
 
   dispatch(SystemTest.hook("export-start"))
-  const {body, path, method} = zealot.inspect.search(program, {
-    format,
-    from,
-    to,
-    poolId,
-    controlMessages: false
-  })
+  const {body, path, method} = zealot.inspect.query(
+    annotateQuery(program, {
+      from,
+      to,
+      poolId
+    }),
+    {
+      format,
+      controlMessages: false
+    }
+  )
   return new Promise<string>((resolve, reject) => {
     http
       .request(

src/js/flows/getZealot.ts

@@ -1,5 +1,5 @@
 import {createFetcher, FetchArgs, Zealot} from "zealot"
-import {ZFetcher, ZReponse} from "../../../zealot/types"
+import {ZFetcher, ZResponse} from "../../../zealot/types"
 import {validateToken} from "../auth0/utils"
 import {BrimWorkspace} from "../brim"
 import ErrorFactory from "../models/ErrorFactory"
@@ -55,11 +55,11 @@ const createBrimFetcher = (dispatch, getState, workspace: BrimWorkspace) => {
       })
     }
 
-    const wrappedStream = async (args: FetchArgs): Promise<ZReponse> => {
+    const wrappedStream = async (args: FetchArgs): Promise<ZResponse> => {
       return stream(await setWorkspaceAuthArgs(workspace, args))
     }
 
-    const wrappedUpload = async (args: FetchArgs): Promise<ZReponse> => {
+    const wrappedUpload = async (args: FetchArgs): Promise<ZResponse> => {
       return upload(await setWorkspaceAuthArgs(workspace, args))
     }