-
Notifications
You must be signed in to change notification settings - Fork 67
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Packet Post Endpoint: Stream status updates (#378)
Posting large pcap files to a space can often take a long time as zeek slowly chugs away. When a pcap is posted to a space via the POST /space/:space/packet endpoint api now provides streaming updates in order to provide clients with an idea of how far along the ingest process is: When a the request is initiated, a zeek subprocess is spawned, and the contents of the posted pcap is piped sequentially into zeek stdin and a pcap index writer. Assuming this all starts with no errors, a 200 ok status messaged is returned once the first zeek log files in temp zeek log file directory have been written to disk. A TaskStart response is transmitted over the response stream. What data that has been written is also transformed to sorted bzng. At this point the space is queryable. On a hardwired two second interval, the server streams back piped json status update payloads that contain: The start timestamp of ingest, the update timestamp, the total size of pcap file and the size of bytes read from the pcap file. From this information a client should be able to approximate the percentage completion of the ingest process (this won't be entirely accurate, however, it doesn't account for the time it will take transform the zeek logs in to time sorted bzng at the end). A TaskEnd message is sent at the completion of ingest. If TaskEnd.Error is not null users will know that an error occurred during ingest. If an error occurs during ingest, the process is aborted and the space is reset. Also: Add api.Stream/JSONPipe functionality to make it easier to read/write piped json files over the wire.
- Loading branch information
Showing
13 changed files
with
514 additions
and
110 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,31 @@ | ||
| package api | ||
|
|
||
| import ( | ||
| "bufio" | ||
| "bytes" | ||
| "io" | ||
| ) | ||
|
|
||
| var sep = []byte("\n\n") | ||
|
|
||
| func NewJSONPipeScanner(r io.Reader) *bufio.Scanner { | ||
| s := bufio.NewScanner(r) | ||
| s.Split(splitJSONPipe) | ||
| return s | ||
| } | ||
|
|
||
| func splitJSONPipe(data []byte, atEOF bool) (advance int, token []byte, err error) { | ||
| if atEOF && len(data) == 0 { | ||
| return 0, nil, nil | ||
| } | ||
| if i := bytes.Index(data, sep); i >= 0 { | ||
| // We have a full newline-terminated line. | ||
| return i + 2, data[0:i], nil | ||
| } | ||
| // If we're at EOF, we have a final, non-terminated line. Return it. | ||
| if atEOF { | ||
| return len(data), data, nil | ||
| } | ||
| // Request more data. | ||
| return 0, nil, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,47 @@ | ||
| package api | ||
|
|
||
| import ( | ||
| "bufio" | ||
| "context" | ||
| "io" | ||
| ) | ||
|
|
||
| type Stream struct { | ||
| scanner *bufio.Scanner | ||
| cancel context.CancelFunc | ||
| } | ||
|
|
||
| func NewStream(s *bufio.Scanner, c context.CancelFunc) *Stream { | ||
| return &Stream{s, c} | ||
| } | ||
|
|
||
| func (s *Stream) end() { | ||
| if s.cancel != nil { | ||
| s.cancel() | ||
| s.cancel = nil | ||
| } | ||
| } | ||
|
|
||
| func (s *Stream) Next() (interface{}, error) { | ||
| if s.scanner.Scan() { | ||
| v, err := unpack(s.scanner.Bytes()) | ||
| if err != nil { | ||
| s.end() | ||
| return nil, err | ||
| } | ||
| end, ok := v.(*TaskEnd) | ||
| if ok { | ||
| if end.Error != nil { | ||
| err = end.Error | ||
| s.end() | ||
| } | ||
| } | ||
| return v, err | ||
| } | ||
| s.end() | ||
| err := s.scanner.Err() | ||
| if err != io.EOF { | ||
| return nil, err | ||
| } | ||
| return nil, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| package api | ||
|
|
||
| import ( | ||
| "encoding/json" | ||
| "fmt" | ||
| ) | ||
|
|
||
| // unpack transforms a piped json stream into the appropriate api response | ||
| // and returns it as an empty interface so that the caller can receive | ||
| // a stream of objects, check their types, and process them accordingly. | ||
| func unpack(b []byte) (interface{}, error) { | ||
| var v interface{} | ||
| err := json.Unmarshal(b, &v) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| object, ok := v.(map[string]interface{}) | ||
| if !ok { | ||
| return nil, fmt.Errorf("bad json object: %s", string(b)) | ||
| } | ||
| which, ok := object["type"] | ||
| if !ok { | ||
| return nil, fmt.Errorf("no type field in search result: %s", string(b)) | ||
| } | ||
| var out interface{} | ||
| switch which { | ||
| case "TaskStart": | ||
| out = &TaskStart{} | ||
| case "TaskEnd": | ||
| out = &TaskEnd{} | ||
| case "SearchRecords": | ||
| out = &SearchRecords{} | ||
| case "SearchWarnings": | ||
| out = &SearchWarnings{} | ||
| case "SearchStats": | ||
| out = &SearchStats{} | ||
| case "SearchEnd": | ||
| out = &SearchEnd{} | ||
| case "PacketPostStatus": | ||
| out = &PacketPostStatus{} | ||
| default: | ||
| return nil, fmt.Errorf("unknown type in results stream: %s", which) | ||
| } | ||
| err = json.Unmarshal(b, out) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return out, nil | ||
| } |
Oops, something went wrong.