diff --git a/Makefile b/Makefile index a14ae4f..b7f7f11 100644 --- a/Makefile +++ b/Makefile @@ -4,7 +4,7 @@ OVERRIDE_FILE=./Docker/config/local_override.json BINARY=./output/cvelociraptor CONFIG_ARGS= --config $(SERVER_CONFIG) --override_file $(OVERRIDE_FILE) CLIENT_CONFIG_ARGS= --config $(CLIENT_CONFIG) --override_file $(OVERRIDE_FILE) -DLV=dlv debug --init ./scripts/dlv.init --build-flags="-tags 'server_vql extras'" ./bin/ -- --debug --debug_filter result_set +DLV=dlv debug --build-flags="-tags 'server_vql extras'" ./bin/ -- WRITEBACK_DIR=/tmp/pool_writebacks/ POOL_NUMBER=20 @@ -31,7 +31,7 @@ dump: $(BINARY) $(CONFIG_ARGS) elastic dump -v dump_persisted: - $(BINARY) $(CONFIG_ARGS) elastic dump --index="persisted" -v --dump_count 2000 + $(BINARY) $(CONFIG_ARGS) elastic dump --index="persisted" -v --org_id O123 debug_gui: $(DLV) $(CONFIG_ARGS) gui -v diff --git a/bin/debug.go b/bin/debug.go index 5526962..e4fec83 100644 --- a/bin/debug.go +++ b/bin/debug.go @@ -22,7 +22,6 @@ import ( "log" "net/http" _ "net/http/pprof" - "regexp" "www.velocidex.com/golang/cloudvelo/services" config_proto "www.velocidex.com/golang/velociraptor/config/proto" @@ -31,9 +30,7 @@ import ( ) var ( - debug_flag = app.Flag("debug", "Enables debug and profile server.").Bool() - debug_regex = app.Flag("debug_filter", "A regex to filter the debug source."). - Default(".").String() + debug_flag = app.Flag("debug", "Enables debug and profile server.").Bool() debug_flag_port = app.Flag("debug_port", "Port for the debug server."). Default("6060").Int64() ) @@ -43,12 +40,7 @@ func initDebugServer(config_obj *config_proto.Config) error { logger := logging.GetLogger(config_obj, &logging.FrontendComponent) logger.Info("Starting debug server on http://127.0.0.1:%v/debug/pprof", *debug_flag_port) - re, err := regexp.Compile("(?i)" + *debug_regex) - if err != nil { - return err - } - - services.SetDebugLogger(config_obj, re) + services.SetDebugLogger(config_obj) // Switch off the debug flag so we do not run this again. (The // GUI runs this function multiple times). diff --git a/config/api.go b/config/api.go index 6502557..59c139d 100644 --- a/config/api.go +++ b/config/api.go @@ -47,14 +47,6 @@ type ElasticConfiguration struct { DedicatedForeman bool `json:"dedicated_foreman"` DedicatedForemanOrg string `json:"dedicated_foreman_org"` ForemanExcludedOrgs []string `json:"foreman_excluded_orgs"` - - // How many rows to collect before flushing into a result set - // packet. Default is 100. - RowsPerResultSet uint64 `json:"rows_per_result_set"` - MaxSizePerPacket uint64 `json:"max_size_per_packet"` - - // Minimum amount of time we cache flow indexes (Default 1) - MinFlowCacheTimeMin int64 `json:"min_flow_cache_time_min"` } // Create a new cloud config object which contains the original diff --git a/go.mod b/go.mod index fc91a58..57b34c3 100644 --- a/go.mod +++ b/go.mod @@ -11,7 +11,6 @@ require ( github.com/alecthomas/assert v1.0.0 github.com/aws/aws-sdk-go v1.55.6 github.com/aws/aws-sdk-go-v2/config v1.27.6 - github.com/codesoap/lineworker v0.2.1 github.com/evanphx/json-patch/v5 v5.9.11 github.com/google/uuid v1.6.0 github.com/magefile/mage v1.15.0 diff --git a/go.sum b/go.sum index d2dc7cc..485d1d2 100644 --- a/go.sum +++ b/go.sum @@ -229,8 +229,6 @@ github.com/clbanning/mxj v1.8.4 h1:HuhwZtbyvyOw+3Z1AowPkU87JkJUSv751ELWaiTpj8I= github.com/clbanning/mxj v1.8.4/go.mod h1:BVjHeAH+rl9rs6f+QIpeRl0tfu10SXn1pUSa5PVGJng= github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw= github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc= -github.com/codesoap/lineworker v0.2.1 h1:YZI1Xamk7/i9J3iPWz7hopttZKwFEn0GXOsF5UeebNI= -github.com/codesoap/lineworker v0.2.1/go.mod h1:aatQ4DVq3bEOUG0O5lAcBhSiwH2BmwmSYQPJjrYCv6w= github.com/coreos/go-oidc/v3 v3.11.0 h1:Ia3MxdwpSw702YW0xgfmP1GVCMA9aEFWu12XUZ3/OtI= github.com/coreos/go-oidc/v3 v3.11.0/go.mod h1:gE3LgjOgFoHi9a4ce4/tJczr0Ai2/BoDhf0r5lltWI0= github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= diff --git a/ingestion/fixtures/TestListDirectory.golden b/ingestion/fixtures/TestListDirectory.golden index 607ac05..aa37dde 100644 --- a/ingestion/fixtures/TestListDirectory.golden +++ b/ingestion/fixtures/TestListDirectory.golden @@ -86,38 +86,6 @@ "doc_type": "vfs", "doc_id": "1cbcacc4b30a75913df3a051a73742ca2bca1963", "timestamp": 1661385600000000000 - }, - { - "timestamp": 1661385600000000000, - "vfs_path": "/clients/C.77ad4285690698d9/artifacts/System.VFS.ListDirectory/F.CEV6I8LHAT83O/Listing.json", - "id": "", - "end_row": 1, - "total_rows": 0, - "type": "rs_metadata" - }, - { - "timestamp": 1661385600000000000, - "vfs_path": "/clients/C.77ad4285690698d9/artifacts/System.VFS.ListDirectory/F.CEV6I8LHAT83O/Stats.json", - "id": "", - "end_row": 1, - "total_rows": 0, - "type": "rs_metadata" - }, - { - "timestamp": 1661385600000000000, - "vfs_path": "/clients/C.77ad4285690698d9/collections/F.CEV6I8LHAT83O/logs.json", - "id": "", - "end_row": 14, - "total_rows": 0, - "type": "rs_metadata" - }, - { - "timestamp": 1661385600000000000, - "vfs_path": "/clients/C.77ad4285690698d9/collections/F.CEV6I8LHAT83O/logs.json", - "id": "", - "end_row": 5, - "total_rows": 0, - "type": "rs_metadata" } ], "System.VFS.ListDirectory vfs": [ diff --git a/result_sets/simple/factory.go b/result_sets/simple/factory.go index 3214b35..85f7302 100644 --- a/result_sets/simple/factory.go +++ b/result_sets/simple/factory.go @@ -6,22 +6,12 @@ import ( "time" "www.velocidex.com/golang/cloudvelo/filestore" - cvelo_services "www.velocidex.com/golang/cloudvelo/services" "www.velocidex.com/golang/velociraptor/file_store/api" "www.velocidex.com/golang/velociraptor/json" "www.velocidex.com/golang/velociraptor/result_sets" "www.velocidex.com/golang/velociraptor/utils" ) -/* - Since we write result sets on the transient index we can not - actually delete anything. - - Therefore we end up writing a new ID to indicate a new result set - and we write a metadata record to point the result set at the latest - ID. -*/ - type ResultSetFactory struct{} func (self ResultSetFactory) NewResultSetWriter( @@ -31,19 +21,7 @@ func (self ResultSetFactory) NewResultSetWriter( completion func(), truncate result_sets.WriteMode) (result_sets.ResultSetWriter, error) { - cvelo_services.Count("NewResultSetWriter") - cloud_config_obj := filestore.GetConfigObj(file_store_factory) - rows_per_result_set := cloud_config_obj.Cloud.RowsPerResultSet - if rows_per_result_set == 0 { - rows_per_result_set = 1000 - } - - max_size_per_packet := cloud_config_obj.Cloud.MaxSizePerPacket - if max_size_per_packet == 0 { - max_size_per_packet = 1024 * 1024 - } - config_obj := cloud_config_obj.VeloConf() new_id := fmt.Sprintf("%v", utils.GetGUID()) @@ -55,7 +33,6 @@ func (self ResultSetFactory) NewResultSetWriter( Timestamp: utils.GetTime().Now().UnixNano(), VFSPath: base_record.VFSPath, ID: new_id, - EndRow: 0, Type: "rs_metadata", } @@ -76,17 +53,12 @@ func (self ResultSetFactory) NewResultSetWriter( } return &ElasticSimpleResultSetWriter{ - org_id: utils.GetOrgId(config_obj), - config_obj: config_obj, - log_path: log_path, - opts: opts, - ctx: context.Background(), - sync: utils.CompareFuncs(completion, utils.SyncCompleter), - version: md.ID, - md: md, - start_row: md.EndRow, - rows_per_result_set: rows_per_result_set, - max_size_per_packet: max_size_per_packet, + org_id: utils.GetOrgId(config_obj), + log_path: log_path, + opts: opts, + ctx: context.Background(), + sync: utils.CompareFuncs(completion, utils.SyncCompleter), + version: md.ID, }, nil } @@ -94,8 +66,6 @@ func (self ResultSetFactory) NewResultSetReader( file_store_factory api.FileStore, log_path api.FSPathSpec) (result_sets.ResultSetReader, error) { - cvelo_services.Count("NewResultSetReader") - ctx := context.Background() cloud_config_obj := filestore.GetConfigObj(file_store_factory) @@ -107,29 +77,10 @@ func (self ResultSetFactory) NewResultSetReader( return nil, utils.NotFoundError } - // This signifies that the result set is incomplete - we can not - // open it. It happens when we abort the writing of the result set - // prematurely. - if existing_md.TotalRows < 0 { - return nil, utils.NotFoundError - } - - base_record := NewSimpleResultSetRecord(log_path, existing_md.ID) - - // Backwards compatibility - should not be needed with newer results - if existing_md.EndRow == 0 { - org_id := filestore.GetOrgId(file_store_factory) - last_rec, err := getLastRecord(ctx, org_id, base_record) - if err == nil { - existing_md.EndRow = last_rec.EndRow - } - } - return &SimpleResultSetReader{ file_store_factory: file_store_factory, log_path: log_path, mtime: time.Unix(0, existing_md.Timestamp), - base_record: base_record, - md: existing_md, + base_record: NewSimpleResultSetRecord(log_path, existing_md.ID), }, nil } diff --git a/result_sets/simple/metadata.go b/result_sets/simple/metadata.go index 19e6846..fe6533c 100644 --- a/result_sets/simple/metadata.go +++ b/result_sets/simple/metadata.go @@ -12,7 +12,6 @@ import ( ) const ( - // Get the latest metadata record with the most recent timestamp. result_md_query = ` { "query": {"bool": {"must": [ @@ -29,8 +28,6 @@ type ResultSetMetadataRecord struct { Timestamp int64 `json:"timestamp"` VFSPath string `json:"vfs_path"` ID string `json:"id"` - EndRow int64 `json:"end_row"` - TotalRows int64 `json:"total_rows"` Type string `json:"type"` } @@ -44,11 +41,6 @@ func GetResultSetMetadata( config_obj *config_proto.Config, log_path api.FSPathSpec) (*ResultSetMetadataRecord, error) { - cvelo_services.Count("GetResultSetMetadata") - cvelo_services.Debug( - cvelo_services.DEBUG_RESULT_SET, - "GetResultSetMetadata: %v", log_path.AsClientPath())() - base_record := NewSimpleResultSetRecord(log_path, "") query := json.Format(result_md_query, base_record.VFSPath) diff --git a/result_sets/simple/reader.go b/result_sets/simple/reader.go index 696f02e..0abe20e 100644 --- a/result_sets/simple/reader.go +++ b/result_sets/simple/reader.go @@ -43,7 +43,6 @@ type SimpleResultSetReader struct { opts result_sets.ResultSetOptions base_record *SimpleResultSetRecord mtime time.Time - md *ResultSetMetadataRecord stacker api.FSPathSpec } @@ -61,11 +60,6 @@ func (self *SimpleResultSetReader) SeekToRow(start int64) error { func (self *SimpleResultSetReader) getPacket( ctx context.Context, row int64) (*SimpleResultSetRecord, error) { - cvelo_services.Count("SimpleResultSet: getPacket") - cvelo_services.Debug( - cvelo_services.DEBUG_RESULT_SET, "getPacket: %v (%v)", - self.log_path, row)() - var artifact_clause, query string if self.base_record.VFSPath != "" { @@ -137,11 +131,6 @@ func (self *SimpleResultSetReader) Rows( break } - // We read enough rows - if self.md.EndRow > 0 && self.row >= self.md.EndRow { - break - } - packet, err := self.getPacket(ctx, self.row) if err != nil { return @@ -201,11 +190,6 @@ func (self *SimpleResultSetReader) JSON( break } - // We read enough rows - if self.md.EndRow > 0 && self.row >= self.md.EndRow { - break - } - packet, err := self.getPacket(ctx, self.row) if err != nil { return @@ -260,21 +244,20 @@ func (self *SimpleResultSetReader) Close() {} // Figure out how many rows are in this collection in total. func (self *SimpleResultSetReader) TotalRows() int64 { - return self.md.EndRow + org_id := filestore.GetOrgId(self.file_store_factory) + last_rec, err := getLastRecord(org_id, self.base_record) + if err != nil { + return -1 + } + + return last_rec.EndRow } -// Deprecated: Current code sets the last row number in another -// metadata document in the EndRow field. This is here for backwards -// compatibility. -func getLastRecord( - ctx context.Context, - org_id string, +func getLastRecord(org_id string, base_record *SimpleResultSetRecord) (*SimpleResultSetRecord, error) { - + ctx := context.Background() var artifact_clause, query string - cvelo_services.Count("SimpleResultSet: getLastRecord") - if base_record.VFSPath != "" { query = json.Format(` {"sort": {"end_row": {"order": "desc"}}, diff --git a/result_sets/simple/transformed.go b/result_sets/simple/transformed.go index 2e6beec..863e648 100644 --- a/result_sets/simple/transformed.go +++ b/result_sets/simple/transformed.go @@ -6,7 +6,6 @@ import ( "time" "github.com/Velocidex/ordereddict" - cvelo_services "www.velocidex.com/golang/cloudvelo/services" config_proto "www.velocidex.com/golang/velociraptor/config/proto" "www.velocidex.com/golang/velociraptor/file_store/api" "www.velocidex.com/golang/velociraptor/result_sets" @@ -24,19 +23,14 @@ func (self ResultSetFactory) NewResultSetReaderWithOptions( log_path api.FSPathSpec, options result_sets.ResultSetOptions) (result_sets.ResultSetReader, error) { - cvelo_services.Count("NewResultSetReaderWithOptions") - - base_reader, err := self.NewResultSetReader(file_store_factory, log_path) + existing_md, err := GetResultSetMetadata(ctx, config_obj, log_path) if err != nil { - // The base result set does not exist. We dont care about any - // transformations. return nil, err } // First do the filtering and then do the sorting. return self.getFilteredReader( - ctx, config_obj, file_store_factory, - log_path, base_reader, options) + ctx, config_obj, file_store_factory, log_path, existing_md.ID, options) } func (self ResultSetFactory) getFilteredReader( @@ -44,14 +38,14 @@ func (self ResultSetFactory) getFilteredReader( config_obj *config_proto.Config, file_store_factory api.FileStore, log_path api.FSPathSpec, - base_reader result_sets.ResultSetReader, + version string, options result_sets.ResultSetOptions) (result_sets.ResultSetReader, error) { // No filter required. if options.FilterColumn == "" || options.FilterRegex == nil { return self.getSortedReader(ctx, config_obj, file_store_factory, - log_path, base_reader, options) + log_path, version, options) } transformed_path := log_path @@ -65,17 +59,35 @@ func (self ResultSetFactory) getFilteredReader( if options.FilterExclude { transformed_path = transformed_path.AddChild("exclude") } + transformed_result_set_record := NewSimpleResultSetRecord( + transformed_path, version) - // Do we have a cached transformed result set? If yes and it is - // newer than the base result set, then just use it. - transformed_reader, err := self.NewResultSetReader(file_store_factory, transformed_path) - if err == nil && transformed_reader.MTime().After(base_reader.MTime()) { - return self.getSortedReader(ctx, config_obj, file_store_factory, - transformed_path, transformed_reader, options) + // Try to open the transformed result set if it is already cached. + log_result_set_record := NewSimpleResultSetRecord(log_path, version) + last_record, err := getLastRecord(config_obj.OrgId, log_result_set_record) + if err != nil { + // Original Result set is not found - just return an empty + // one. + return self.NewResultSetReader(file_store_factory, log_path) + } + + transformed_last_record, err := getLastRecord( + config_obj.OrgId, transformed_result_set_record) + if err == nil && + // Existing result is still valid, lets use it. + transformed_last_record.Timestamp > last_record.Timestamp { + return self.getSortedReader(ctx, config_obj, + file_store_factory, transformed_path, version, options) + } + + // Nope - we have to build the new cache from the original table. + reader, err := self.NewResultSetReader(file_store_factory, log_path) + if err != nil { + return nil, err } + defer reader.Close() - base_reader, err = simple.WrapReaderForRange( - base_reader, options.StartIdx, options.EndIdx) + reader, err = simple.WrapReaderForRange(reader, options.StartIdx, options.EndIdx) if err != nil { return nil, err } @@ -92,7 +104,7 @@ func (self ResultSetFactory) getFilteredReader( defer sub_cancel() // Filter the table with the regex - row_chan := base_reader.Rows(sub_ctx) + row_chan := reader.Rows(sub_ctx) outer: for { @@ -117,11 +129,6 @@ outer: } } - if utils.IsCtxDone(sub_ctx) { - AbortResultSet(writer) - return nil, utils.IOError - } - // Flush all the writes back writer.Close() @@ -130,15 +137,8 @@ outer: options.StartIdx = 0 options.EndIdx = 0 - // Reopen the result set - transformed_reader, err = self.NewResultSetReader( - file_store_factory, transformed_path) - if err != nil { - return nil, err - } - return self.getSortedReader(ctx, config_obj, file_store_factory, - transformed_path, transformed_reader, options) + transformed_path, version, options) } func (self ResultSetFactory) getSortedReader( @@ -146,12 +146,16 @@ func (self ResultSetFactory) getSortedReader( config_obj *config_proto.Config, file_store_factory api.FileStore, log_path api.FSPathSpec, - base_reader result_sets.ResultSetReader, + version string, options result_sets.ResultSetOptions) (result_sets.ResultSetReader, error) { // No sorting required. if options.SortColumn == "" { - return simple.WrapReaderForRange(base_reader, options.StartIdx, options.EndIdx) + reader, err := self.NewResultSetReader(file_store_factory, log_path) + if err != nil { + return nil, err + } + return simple.WrapReaderForRange(reader, options.StartIdx, options.EndIdx) } transformed_path := log_path @@ -168,19 +172,44 @@ func (self ResultSetFactory) getSortedReader( "sorted", options.SortColumn, "desc") } - // Do we have a cached transformed result set? If yes and it is - // newer than the base result set, then just use it. - transformed_reader, err := self.NewResultSetReader(file_store_factory, transformed_path) - if err == nil && transformed_reader.MTime().After(base_reader.MTime()) { - return simple.WrapReaderForRange(transformed_reader, options.StartIdx, options.EndIdx) + stacker_path := transformed_path.AddChild("stack") + + transformed_result_set_record := NewSimpleResultSetRecord( + transformed_path, version) + log_result_set_record := NewSimpleResultSetRecord(transformed_path, version) + + // Try to open the transformed result set if it is already cached. + last_record, err := getLastRecord( + config_obj.OrgId, log_result_set_record) + if err == nil { + res, err := self.NewResultSetReader(file_store_factory, transformed_path) + if err == nil { + res.SetStacker(stacker_path) + } + return res, err } - stacker_path := transformed_path.AddChild("stack") + transformed_last_record, err := getLastRecord( + config_obj.OrgId, transformed_result_set_record) + if err == nil && + // Existing result is still valid, lets use it. + transformed_last_record.Timestamp > last_record.Timestamp { + res, err := self.NewResultSetReader(file_store_factory, transformed_path) + if err == nil { + res.SetStacker(stacker_path) + } + return res, err + } // Nope - we have to build the new cache from the original table. scope := vql_subsystem.MakeScope() + reader, err := self.NewResultSetReader(file_store_factory, log_path) + if err != nil { + return nil, err + } + defer reader.Close() - reader, err := simple.WrapReaderForRange(base_reader, options.StartIdx, options.EndIdx) + reader, err = simple.WrapReaderForRange(reader, options.StartIdx, options.EndIdx) if err != nil { return nil, err } @@ -215,7 +244,7 @@ func (self ResultSetFactory) getSortedReader( go func() { defer close(sorter_input_chan) - row_chan := reader.Rows(sub_ctx) + row_chan := reader.Rows(ctx) for { select { case <-sub_ctx.Done(): @@ -237,15 +266,9 @@ func (self ResultSetFactory) getSortedReader( } } - if utils.IsCtxDone(sub_ctx) { - AbortResultSet(writer) - return nil, utils.IOError - } - // Close synchronously to flush the data writer.Close() - // Reopen the result set and return it. result, err := self.NewResultSetReader(file_store_factory, transformed_path) if err != nil { return nil, err @@ -265,12 +288,3 @@ func getExpiry(config_obj *config_proto.Config) time.Duration { return 10 * time.Minute } - -// Abort writing the result set by setting the TotalRows to -1 to -// signal this result set is incomplete. -func AbortResultSet(writer result_sets.ResultSetWriter) { - rs_writer, ok := writer.(*ElasticSimpleResultSetWriter) - if ok { - rs_writer.Abort() - } -} diff --git a/result_sets/simple/writer.go b/result_sets/simple/writer.go index 4a20d21..20974f5 100644 --- a/result_sets/simple/writer.go +++ b/result_sets/simple/writer.go @@ -3,11 +3,11 @@ package simple import ( "context" "errors" + "strconv" "github.com/Velocidex/ordereddict" "www.velocidex.com/golang/cloudvelo/services" cvelo_services "www.velocidex.com/golang/cloudvelo/services" - config_proto "www.velocidex.com/golang/velociraptor/config/proto" "www.velocidex.com/golang/velociraptor/file_store/api" "www.velocidex.com/golang/velociraptor/json" "www.velocidex.com/golang/velociraptor/utils" @@ -28,17 +28,12 @@ type ElasticSimpleResultSetWriter struct { // and can be expensive). truncated bool - ctx context.Context - config_obj *config_proto.Config + ctx context.Context // If this is set writes will be syncrounous sync bool - md *ResultSetMetadataRecord - - version string - rows_per_result_set uint64 - max_size_per_packet uint64 + version string } // Not currently implemented but in future will be used to update @@ -47,31 +42,9 @@ func (self *ElasticSimpleResultSetWriter) Update(uint64, *ordereddict.Dict) erro return errors.New("Updating result sets is not implemented yet.") } -func (self *ElasticSimpleResultSetWriter) Abort() { - self.md.TotalRows = -1 - self.md.EndRow = -1 - self.Close() -} - func (self *ElasticSimpleResultSetWriter) WriteJSONL( serialized []byte, total_rows uint64) { - // Valid JSONL should be followed by \n already - self.buff = append(self.buff, serialized...) - self.buffered_rows += int(total_rows) - - // Flush depending on the total size of the buffer. If the rows - // are large, we try to keep document size under 1mb. - if uint64(self.buffered_rows) > self.rows_per_result_set || - uint64(len(self.buff)) > self.max_size_per_packet { - self.Flush() - } -} - -// Write the JSONL record into a single document. -func (self *ElasticSimpleResultSetWriter) writeJSONL( - serialized []byte, total_rows uint64) { - record := NewSimpleResultSetRecord(self.log_path, self.version) record.JSONData = string(serialized) record.StartRow = self.start_row @@ -81,23 +54,17 @@ func (self *ElasticSimpleResultSetWriter) writeJSONL( record.Type = "result_set" self.start_row = record.EndRow - self.md.EndRow = record.EndRow - record.TotalRows = uint64(self.start_row) if self.sync { - err := services.SetElasticIndex( + services.SetElasticIndex( self.ctx, self.org_id, "transient", services.DocIdRandom, record) - if err != nil { - self.Abort() - } - return + } else { + services.SetElasticIndexAsync( + self.org_id, "transient", services.DocIdRandom, + cvelo_services.BulkUpdateCreate, record) } - - services.SetElasticIndexAsync( - self.org_id, "transient", services.DocIdRandom, - cvelo_services.BulkUpdateCreate, record) } func (self *ElasticSimpleResultSetWriter) Write(row *ordereddict.Dict) { @@ -110,10 +77,7 @@ func (self *ElasticSimpleResultSetWriter) Write(row *ordereddict.Dict) { self.buff = append(self.buff, '\n') self.buffered_rows++ - // Flush depending on the total size of the buffer. If the rows - // are large, we try to keep document size under 1mb. - if uint64(self.buffered_rows) > self.rows_per_result_set || - uint64(len(self.buff)) > self.max_size_per_packet { + if self.buffered_rows > 100 { self.Flush() } } @@ -127,18 +91,59 @@ func (self *ElasticSimpleResultSetWriter) SetStartRow(start_row int64) error { return nil } +const getLargestRowId = ` +{ + "query": { + "bool": { + "must": [ + {"match": {"type": "result_set"}}, + {"match": {"id": %q}}, + {"match": {"vfs_path": %q}} + ]} + }, + "size": 0, + "aggs": { + "genres": { + "max": {"field": "end_row"} + } + } +} +` + +func (self *ElasticSimpleResultSetWriter) getLastRow() error { + query := json.Format(getLargestRowId, + self.version, self.log_path.AsClientPath()) + + hits, err := services.QueryElasticAggregations( + self.ctx, self.org_id, "transient", query) + + if err != nil { + return err + } + + for _, hit := range hits { + end_row, err := strconv.ParseInt(hit, 10, 64) + if err == nil { + self.start_row = end_row + } + self.truncated = true + } + return nil +} + func (self *ElasticSimpleResultSetWriter) Flush() { if self.buffered_rows == 0 { return } - self.writeJSONL(self.buff, uint64(self.buffered_rows)) + if !self.truncated { + self.getLastRow() + } + + self.WriteJSONL(self.buff, uint64(self.buffered_rows)) self.buff = nil self.buffered_rows = 0 - // Write a newer version of the MD record. - _ = SetResultSetMetadata(self.ctx, self.config_obj, self.log_path, self.md) - // Make sure the results are visible immediately cvelo_services.FlushIndex(self.ctx, self.org_id, "transient") diff --git a/schema/api/clients.go b/schema/api/clients.go index 4c86a7d..0e0922a 100644 --- a/schema/api/clients.go +++ b/schema/api/clients.go @@ -76,8 +76,6 @@ func GetMultipleClients( config_obj *config_proto.Config, client_ids []string) ([]*ClientRecord, error) { - cvelo_services.Count("GetMultipleClients") - terms := make([]string, 0, 2*len(client_ids)) for _, i := range client_ids { terms = append(terms, i) diff --git a/schema/api/collections.go b/schema/api/collections.go index a9ba442..68a13b0 100644 --- a/schema/api/collections.go +++ b/schema/api/collections.go @@ -38,10 +38,6 @@ func (self *ArtifactCollectorRecord) ToProto() ( result.ClientId = self.ClientId } - if result.SessionId == "" { - result.SessionId = self.SessionId - } - return result, nil } diff --git a/scripts/dlv.init b/scripts/dlv.init deleted file mode 100644 index e3d0d02..0000000 --- a/scripts/dlv.init +++ /dev/null @@ -1 +0,0 @@ -source scripts/dlv.star \ No newline at end of file diff --git a/scripts/dlv.star b/scripts/dlv.star deleted file mode 100644 index 0665d2d..0000000 --- a/scripts/dlv.star +++ /dev/null @@ -1,332 +0,0 @@ -"""This is a helper script for debugging Velociraptor. - -You can load it into your dlv session with: - -``` -$ make debug -dlv debug --wd=. --build-flags="-tags 'server_vql extras'" ./bin/ -- frontend --disable-panic-guard -v --debug -Type 'help' for list of commands. -(dlv) source scripts/dlv.star -Loading Velociraptor extenstions -(dlv) -``` - -This debugger extension has the following benefits: -- pp command pretty prints structs: - 1. special support for Velociraptor types like ordereddict.Dict, Scope etc. - 2. Hide unexported fields by default (use ppl to see a longer dump). - 3. Support protobuf by hiding internal fields. - 4. Coloring output to make it easier to see - -- Also some helpful commands - -""" - -# Set up the color theme -reset_code = "\033[0m" -list_line_color = "\x1b[34m" -keyword_color = "\x1b[94m" -string_color = "\x1b[91m" -number_color = "\x1b[37m" -comment_color = "\x1b[92m" -arrow_color = "\x1b[93m" -tab_color = "\x1b[90m" - -proto_fields = ["state", "sizeCache", "unknownFields"] - -LongOptions = { - "include_private": True, - "include_defaults": True, - "shorten_strings": False, -} - -ShortOptions = { - "include_private": False, - "include_defaults": False, - "shorten_strings": True, -} - -def set_config(): - dlv_command("config source-list-line-color " + list_line_color) - dlv_command("config source-list-keyword-color " + keyword_color) - dlv_command("config source-list-string-color " + string_color) - dlv_command("config source-list-number-color " + number_color) - dlv_command("config source-list-comment-color " + comment_color) - dlv_command("config source-list-arrow-color " + arrow_color) - dlv_command("config source-list-tab-color " + tab_color) - dlv_command("config max-variable-recurse 5") - -def debug_Variable(x): - print(repr(x)) - for k in dir(x): - if k == "Value": - continue - print("Variable", k, repr(getattr(x, k))) - -def get_lines(loc, before, after): - return "\n".join(read_file(loc.File).splitlines()[loc.Line+before:loc.Line+after]) - -def get_var(x, name): - for c in x.Children: - if c.Name == name: - return c - return None - -def command_gsl(filter): - """Iterate over all goroutines and print the current line they are at""" - result = "" - gs = goroutines().Goroutines - gs = sorted(gs, key= lambda x: x.ID) - - for g in gs: - loc = g.UserCurrentLoc - cur_line = get_lines(loc, 0, 1) - if not filter in cur_line: - continue - - line = get_lines(loc, -4, -1) + "\n------->" + get_lines(loc, 0, 4) - result += "%d:\t%s:%d\t%s\n" % (g.ID, loc.File, loc.Line, line) - print(result) - -def strip_prefix(value, prefix): - if value.startswith(prefix): - return value[len(prefix):] - return value - -def format(value): - return strip_prefix(repr(value), "interface {}") - -def format_dict(x, indent="", options=LongOptions): - if x.Kind == "interface" and x.Children: - x = x.Children[0] - - if x.Kind == "ptr" and x.Children: - x = x.Children[0] - - items = get_var(x, "items") - if not items: - return indent + "%s%s\n" % (comment_color, reset_code) - - result = "" - for idx, item in enumerate(items.Children): - key = item.Value.Key - value = get_var(item, "Value") - - result += indent + " %s%d%s %s%s%s: %s\n" % ( - number_color, idx, reset_code, - keyword_color, key, reset_code, - format_type(value, options=options, indent=indent + " ")) - return result - -def format_type(x, indent="", options=LongOptions): - if x.Kind == "interface" and x.Children: - x = x.Children[0] - - if x.Kind == "ptr" and x.Children: - x = x.Children[0] - - if x.Kind == "chan" and x.Children: - closed = get_var(x, "closed") - res = "%s%s%s" % (comment_color, x.Type, reset_code) - if closed.Value: - res += " (closed) " - return res - - if x.Kind == "string": - res = str(x.Value) - if options["shorten_strings"]: - lines = res.split("\n") - if len(lines) > 1: - res = lines[0] + "%s ...%s" % (tab_color, reset_code) - - if len(res) > 100: - res = res[:100] + "%s ...%s" % (tab_color, reset_code) - return res - - if x.Unreadable: - return "%s%s" % (comment_color, reset_code) - - if x.Len > 0 and len(x.Children) == 0: - return "%s%s" % (comment_color, reset_code) - - var = x.Value - - if x.Type == "time.Time": - return str(var) - - if x.Type == "void": - return "%snil%s" % (comment_color, reset_code) - - if len(x.Children) == 0: - return str(var) - - if x.Type == "github.com/Velocidex/ordereddict.Dict": - result = indent + "%sDict%s\n" % (string_color, reset_code) - result += format_dict(x, indent=indent, options=options) - return result.rstrip() - - if x.Type == "www.velocidex.com/golang/vfilter/scope.Scope": - result = indent + "Scope (%d)\n" % var.id - vars = get_var(x, "vars") - for idx, v in enumerate(vars.Children): - result += indent + "%sVar %s%s\n" % (string_color, idx, reset_code) - result += format_dict(v, indent=indent + " ", - options=options) - - return result.rstrip() - - if x.Type == "regexp.Regexp": - result = indent + "%sregexp.Regexp%s: %s" % ( - comment_color, reset_code, repr(var.expr)) - return result - - if x.Kind == "slice": - result = "%s%s%s len %d\n" % ( - string_color, x.Type, reset_code, len(x.Children)) - - for idx, v in enumerate(x.Children): - result += indent + "%s%s%s %s\n" % ( - number_color, idx, reset_code, - format_type(v, indent=indent+" ", options=options)) - - return result.rstrip() - - if x.Kind == "struct": - # Format protobufs specially by dropping useless fields. - is_proto = False - omitted_fields = False - result = " %s%s%s {\n" % (string_color, x.Type, reset_code) - for c in x.Children: - if not options["include_defaults"] and is_default(c): - omitted_fields = True - continue - - if not options["include_private"] and is_private(c): - omitted_fields = True - continue - - if c.Name == "state": - is_proto = True - - if is_proto and (c.Name in proto_fields or is_default(c)): - omitted_fields = True - continue - - result += indent + " %s%s%s: %s\n" % ( - keyword_color, c.Name, reset_code, - format_type(c, indent=indent + " ", options=options)) - - if omitted_fields: - result += indent + "%s ... omitted_fields%s\n" % ( - tab_color, reset_code) - - return result.rstrip()+ "\n" + indent + "}" - - if x.Kind == "map": - result = "%s%s%s {\n" % (string_color, x.Type, reset_code) - k = "" - for idx, c in enumerate(x.Children): - if idx % 2 == 0: - k = format(c.Value) - continue - - result += indent + " %s%s%s: %s\n" % ( - keyword_color, k, reset_code, - format_type(c, indent=indent + " ")) - return result.rstrip()+ "\n" + indent + "}" - - return indent + str(var) - -def is_private(x): - return x.Name[0] == x.Name[0].lower() - -def is_default(x): - if (x.Kind == "ptr" or x.Kind == "interface") and len(x.Children) > 0: - x = x.Children[0] - - if x.Type == "void": - return True - - if x.Type == "string" and x.Value == "": - return True - - if (x.Kind == "struct" or x.Kind == "slice") and len(x.Children) == 0: - return True - - if (x.Kind == "uint" or x.Kind == "int") and x.Value == 0: - return True - - if (x.Kind == "float32" or x.Kind == "float64") and x.Value == 0.0: - return True - - if x.Kind == "bool" and not x.Value: - return True - - return False - -def command_ppl(args): - """Pretty print local variables. - -This supports some internal Velociraptor types: -* ordereddict.Dict -* vfilter.Scope -* Protocol buffers - -better display for structs - -Example: -> ppl item - """ - result = "" - x = eval(None, args).Variable - print(format_type(x)) - -def command_pp(args): - """Concise Pretty print local variables. - -This supports some internal Velociraptor types: -* ordereddict.Dict -* vfilter.Scope -* Protocol buffers -* regex.Regexp - -better display for structs - -Example: -> pp item - """ - result = "" - x = eval(None, args).Variable - print(format_type(x, options=ShortOptions)) - - -def command_btt(context=0): - """Print decorated backtrace (enriched bt). - -This adds a listing of each call site in the backtrace. - """ - cur_goroutine = state().State.SelectedGoroutine - st = stacktrace(cur_goroutine.ID, 5) - for idx, f in enumerate(st.Locations): - if "go/src/runtime" in f.File: - continue - - print("%d %s%s%s\n\t\t\t%s%s%s %s%d%s" %( idx, - comment_color, f.Function.Name_, reset_code, - string_color, f.File, reset_code, - number_color, f.Line, reset_code)) - if context > 0: - print(keyword_color + - get_lines(f, -context,-1) + "\n*" + get_lines(f, -1, context) + - reset_code) - -def command_rb(args): - """Rebuild and restart""" - print("Rebuilding") - dlv_command("rebuild") - dlv_command("restart") - dlv_command("continue") - -def main(): - set_config() - print("Loading Velociraptor extenstions") diff --git a/services/client_info/client_info.go b/services/client_info/client_info.go index 7375b14..1d967b9 100644 --- a/services/client_info/client_info.go +++ b/services/client_info/client_info.go @@ -7,7 +7,6 @@ import ( "www.velocidex.com/golang/cloudvelo/schema/api" cvelo_services "www.velocidex.com/golang/cloudvelo/services" - actions_proto "www.velocidex.com/golang/velociraptor/actions/proto" config_proto "www.velocidex.com/golang/velociraptor/config/proto" crypto_proto "www.velocidex.com/golang/velociraptor/crypto/proto" "www.velocidex.com/golang/velociraptor/logging" @@ -77,8 +76,6 @@ func (self *ClientInfoBase) Modify(ctx context.Context, client_id string, func (self *ClientInfoBase) ListClients(ctx context.Context) <-chan string { output_chan := make(chan string) - cvelo_services.Count("ListClients") - go func() { defer close(output_chan) @@ -103,9 +100,6 @@ func (self *ClientInfoBase) ListClients(ctx context.Context) <-chan string { func (self *ClientInfoBase) Set( ctx context.Context, client_info *services.ClientInfo) error { - - cvelo_services.Count("SetClient") - now := utils.GetTime().Now().Unix() return cvelo_services.SetElasticIndex(ctx, @@ -138,33 +132,17 @@ func (self ClientInfoBase) Get( ctx context.Context, client_id string) ( *services.ClientInfo, error) { - cvelo_services.Count("GetClient") - - indexer, err := services.GetIndexer(self.config_obj) + hits, err := api.GetMultipleClients( + ctx, self.config_obj, []string{client_id}) if err != nil { return nil, err } - client_info, err := indexer.FastGetApiClient( - ctx, self.config_obj, client_id) - if err != nil { - return nil, err + if len(hits) == 0 { + return nil, errors.New("Client ID not found") } - return &services.ClientInfo{actions_proto.ClientInfo{ - ClientId: client_info.ClientId, - Hostname: client_info.OsInfo.Hostname, - Fqdn: client_info.OsInfo.Fqdn, - System: client_info.OsInfo.System, - Release: client_info.OsInfo.Release, - Architecture: client_info.OsInfo.Machine, - MacAddresses: client_info.OsInfo.MacAddresses, - FirstSeenAt: client_info.FirstSeenAt * 1000, - Ping: client_info.LastSeenAt * 1000, - LastHuntTimestamp: client_info.LastHuntTimestamp, - LastEventTableVersion: client_info.LastEventTableVersion, - BuildTime: client_info.AgentInformation.BuildTime, - }}, nil + return api.ToClientInfo(hits[0]), nil } func (self ClientInfoBase) GetStats( diff --git a/services/elasticsearch.go b/services/elasticsearch.go index 03adf57..fa0e251 100644 --- a/services/elasticsearch.go +++ b/services/elasticsearch.go @@ -9,7 +9,6 @@ import ( "fmt" "io/ioutil" "os" - "regexp" "strings" "sync" "time" @@ -45,10 +44,6 @@ const ( DocIdRandom = "" PrimaryOpenSearch OpenSearchClusterOptions = iota + 1 SecondaryOpenSearch - - // Levels of debug to match with debug_filter regexp. - DEBUG_ELASTIC = "ELASTIC" - DEBUG_RESULT_SET = "RESULT_SET" ) var ( @@ -60,8 +55,7 @@ var ( TRUE = true - logger_filter *regexp.Regexp - logger *logging.LogContext + logger *logging.LogContext bulk_indexer *BulkIndexer @@ -71,15 +65,10 @@ var ( // The logger is normally installed in the start up sequence with // SetDebugLogger() below. -func Debug(source string, format string, args ...interface{}) func() { - mu.Lock() - defer mu.Unlock() - - should_log := logger != nil && logger_filter.MatchString(source) - +func Debug(format string, args ...interface{}) func() { start := time.Now() return func() { - if should_log { + if logger != nil { args = append(args, time.Now().Sub(start)) logger.Debug(format+" in %v", args...) } @@ -148,7 +137,7 @@ func DeleteDocument( defer Instrument("DeleteDocument")() - defer Debug(DEBUG_ELASTIC, "DeleteDocument %v", id)() + defer Debug("DeleteDocument %v", id)() client, err := GetElasticClient(org_id) if err != nil { return err @@ -223,7 +212,7 @@ func FlushIndex( func UpdateIndex( ctx context.Context, org_id, index, id string, query string) error { defer Instrument("UpdateIndex")() - defer Debug(DEBUG_ELASTIC, "UpdateIndex %v %v", index, id)() + defer Debug("UpdateIndex %v %v", index, id)() return retry(func() error { err := _UpdateIndex(ctx, org_id, index, id, query) return err @@ -333,7 +322,7 @@ func PutTemplate( func SetElasticIndexAsync(org_id, index, id string, action BulkUpdateType, record interface{}) error { - defer Debug(DEBUG_ELASTIC, "SetElasticIndexAsync %v %v", index, id)() + defer Debug("SetElasticIndexAsync %v %v", index, id)() mu.Lock() l_bulk_indexer := bulk_indexer @@ -362,7 +351,7 @@ func SetElasticIndexAsync(org_id, index, id string, func SetElasticIndex(ctx context.Context, org_id, index, id string, record interface{}) error { defer Instrument("SetElasticIndex")() - defer Debug(DEBUG_ELASTIC, "SetElasticIndex %v %v", index, id)() + defer Debug("SetElasticIndex %v %v", index, id)() return retry(func() error { return _SetElasticIndex(ctx, org_id, index, id, record) @@ -445,7 +434,7 @@ type _ElasticResponse struct { // Gets a single elastic record by id. func GetElasticRecordByQuery( ctx context.Context, org_id, index_suffix, query string) (json.RawMessage, error) { - defer Debug(DEBUG_ELASTIC, "GetElasticRecordByQuery %v %v", index_suffix, query)() + defer Debug("GetElasticRecordByQuery %v %v", index_suffix, query)() defer Instrument("GetElasticRecordByQuery")() client, err := GetElasticClient(org_id) @@ -498,7 +487,7 @@ func GetElasticRecordByQuery( // Gets a single elastic record by id. func GetElasticRecord( ctx context.Context, org_id, index, id string) (json.RawMessage, error) { - defer Debug(DEBUG_ELASTIC, "GetElasticRecord %v %v", index, id)() + defer Debug("GetElasticRecord %v %v", index, id)() defer Instrument("GetElasticRecord")() client, err := GetElasticClient(org_id) @@ -571,9 +560,9 @@ func GetMultipleElasticRecords( } if len(ids) > 4 { - defer Debug(DEBUG_ELASTIC, "GetMultipleElasticRecords %v %v ...", index, ids[:4])() + defer Debug("GetMultipleElasticRecords %v %v ...", index, ids[:4])() } else { - defer Debug(DEBUG_ELASTIC, "GetMultipleElasticRecords %v %v", index, ids)() + defer Debug("GetMultipleElasticRecords %v %v", index, ids)() } client, err := GetElasticClient(org_id) @@ -647,7 +636,7 @@ func QueryChan( org_id, index, query, sort_field string) ( chan json.RawMessage, error) { - defer Debug(DEBUG_ELASTIC, "QueryChan %v", index)() + defer Debug("QueryChan %v", index)() output_chan := make(chan json.RawMessage) @@ -712,12 +701,6 @@ func QueryChan( } } - // The last page is short so this means it is the last - // page. - if len(part) < page_size { - break - } - // Form the next query using the search_after value. part_query := json.Format(` {"sort":[{%q: "asc"}], "size":%q,"search_after": [%q],`, @@ -773,7 +756,7 @@ func QueryElasticAggregations( ctx context.Context, org_id, index, query string) ([]string, error) { defer Instrument("QueryElasticAggregations")() - defer Debug(DEBUG_ELASTIC, "QueryElasticAggregations %v", index)() + defer Debug("QueryElasticAggregations %v", index)() es, err := GetElasticClient(org_id) if err != nil { @@ -835,7 +818,7 @@ func QueryElasticRaw( org_id, index, query string) ([]json.RawMessage, int, error) { defer Instrument("QueryElasticRaw")() - defer Debug(DEBUG_ELASTIC, "QueryElasticRaw %v", query)() + defer Debug("QueryElasticRaw %v", index)() es, err := GetElasticClient(org_id) if err != nil { @@ -1067,11 +1050,10 @@ func SetElasticClient(clientKey OpenSearchClusterOptions, c *opensearch.Client) elasticClients[clientKey] = c } -func SetDebugLogger(config_obj *config_proto.Config, filter *regexp.Regexp) { +func SetDebugLogger(config_obj *config_proto.Config) { mu.Lock() defer mu.Unlock() - logger_filter = filter logger = logging.GetLogger(config_obj, &logging.FrontendComponent) } @@ -1181,7 +1163,7 @@ func makeReadElasticError(data []byte) error { // Now that indexes are created from the templates, a missing // index means that it was not written to yet. //return os.ErrNotExist - Debug(DEBUG_ELASTIC, "ElasticError: %v\n", response)() + Debug("ElasticError: %v\n", response)() return nil } diff --git a/services/exports/exports.go b/services/exports/exports.go index 9b15d93..b24efe2 100644 --- a/services/exports/exports.go +++ b/services/exports/exports.go @@ -91,8 +91,6 @@ func (self *ExportManager) GetAvailableDownloadFiles( config_obj *config_proto.Config, opts services.ContainerOptions) (*api_proto.AvailableDownloads, error) { - cvelo_services.Count("ExportManager: GetAvailableDownloadFiles") - filter := "" switch opts.Type { case services.NotebookExport: diff --git a/services/hunt_dispatcher/flows.go b/services/hunt_dispatcher/flows.go index eaa9328..ece9048 100644 --- a/services/hunt_dispatcher/flows.go +++ b/services/hunt_dispatcher/flows.go @@ -6,9 +6,7 @@ import ( "io" "time" - "github.com/codesoap/lineworker" cvelo_services "www.velocidex.com/golang/cloudvelo/services" - "www.velocidex.com/golang/cloudvelo/services/indexing" api_proto "www.velocidex.com/golang/velociraptor/api/proto" config_proto "www.velocidex.com/golang/velociraptor/config/proto" "www.velocidex.com/golang/velociraptor/file_store" @@ -21,12 +19,6 @@ import ( "www.velocidex.com/golang/vfilter" ) -type job_t struct { - ClientId string - FlowId string - Details *api_proto.FlowDetails -} - const ( getHuntsFlowsQuery = `{ "from": %q, "query": { @@ -63,64 +55,38 @@ type HuntFlowEntry struct { DocType string `json:"doc_type"` } -func (self *HuntDispatcher) shouldRebuildIndex( - ctx context.Context, - config_obj *config_proto.Config, - hunt_id string) bool { +func (self *HuntDispatcher) syncFlowTables( + ctx context.Context, config_obj *config_proto.Config, + hunt_id string) error { file_store_factory := file_store.GetFileStore(config_obj) hunt_path_manager := paths.NewHuntPathManager(hunt_id) table_to_query := hunt_path_manager.EnrichedClients() - rs_reader, err := result_sets.NewResultSetReader( - file_store_factory, table_to_query) - if err != nil { - return true - } - - now := utils.GetTime().Now() - - // If the index is too old, then rebuild it anyway. TODO: This can - // be relaxed when the indexing gets more stable. - if now.Sub(rs_reader.MTime()) > time.Hour*12 { - return true - } - // First get the latest hunt_flow document. This indicates the // last time the hunt was seen. hit, err := cvelo_services.GetElasticRecordByQuery(ctx, config_obj.OrgId, cvelo_services.TRANSIENT, json.Format( getLatestHuntFlowForHuntId, hunt_id)) - if err != nil || len(hit) == 0 { - return true - } + if err == nil && len(hit) > 1 { + entry := &HuntFlowEntry{} + err = json.Unmarshal(hit, entry) + if err == nil { + last_modified := time.Unix(entry.Timestamp, 0) - entry := &HuntFlowEntry{} - err = json.Unmarshal(hit, entry) - if err != nil { - return true - } + // Now check the last modified time of the result set. + rs_reader, err := result_sets.NewResultSetReader( + file_store_factory, table_to_query) + if err == nil && rs_reader.MTime().After(last_modified) { - // This is the last modified time of any flow in the hunt. - last_modified := time.Unix(entry.Timestamp, 0) + // Skip the update if the result set is newer than the + // last_modified record. + return nil + } - // Now check the last modified time of the result set. - if rs_reader.MTime().Before(last_modified) { - return true + } } - // Skip the update if the result set is newer than the - // last_modified record. - return false -} - -func (self *HuntDispatcher) syncFlowTables( - ctx context.Context, - config_obj *config_proto.Config, - hunt_id string) error { - - defer cvelo_services.Summarize("HuntDispatcher: syncFlowTables")() - count := 0 seen := make(map[string]bool) @@ -129,19 +95,6 @@ func (self *HuntDispatcher) syncFlowTables( return err } - err = indexing.PopulateClientInfoCache(ctx, config_obj) - if err != nil { - return err - } - - // Needs to be immediately available because we will query it - // right away. - defer cvelo_services.FlushIndex(ctx, self.config_obj.OrgId, "transient") - - file_store_factory := file_store.GetFileStore(config_obj) - hunt_path_manager := paths.NewHuntPathManager(hunt_id) - table_to_query := hunt_path_manager.EnrichedClients() - rs_writer, err := result_sets.NewResultSetWriter(file_store_factory, table_to_query, json.DefaultEncOpts(), utils.SyncCompleter, result_sets.TruncateMode) @@ -158,49 +111,26 @@ func (self *HuntDispatcher) syncFlowTables( return err } - // Get the flow details in a worker pool - pool := lineworker.NewWorkerPool(30, func(entry *job_t) (*job_t, error) { - flow, err := laucher_manager.GetFlowDetails( - ctx, config_obj, services.GetFlowOptions{}, - entry.ClientId, entry.FlowId) - if err == nil { - entry.Details = flow + for hit := range hits { + entry := &HuntFlowEntry{} + err = json.Unmarshal(hit, entry) + if err != nil { + continue } - return entry, err - }) - go func() { - defer pool.Stop() - for hit := range hits { - entry := &HuntFlowEntry{} - err = json.Unmarshal(hit, entry) - if err != nil { - continue - } - - key := entry.FlowId + entry.ClientId - if seen[key] { - continue - } - seen[key] = true - pool.Process(&job_t{ - ClientId: entry.ClientId, - FlowId: entry.FlowId, - }) + if seen[entry.FlowId+entry.ClientId] { + continue } - }() + seen[entry.FlowId+entry.ClientId] = true - for { - entry, err := pool.Next() - if err == lineworker.EOS { - break - } + flow, err := laucher_manager.GetFlowDetails( + ctx, config_obj, services.GetFlowOptions{}, + entry.ClientId, entry.FlowId) if err != nil { continue } - count++ - flow := entry.Details + count++ rs_writer.WriteJSONL([]byte( json.Format(`{"ClientId": %q, "Hostname": %q, "FlowId": %q, "StartedTime": %q, "State": %q, "Duration": %q, "TotalBytes": %q, "TotalRows": %q} `, @@ -214,6 +144,10 @@ func (self *HuntDispatcher) syncFlowTables( flow.Context.TotalCollectedRows)), 1) } + // Needs to be immediately available because we will query it + // right away. + cvelo_services.FlushIndex(ctx, self.config_obj.OrgId, "transient") + return nil } @@ -225,15 +159,7 @@ func (self HuntDispatcher) GetFlows( output_chan := make(chan *api_proto.FlowDetails) - if self.shouldRebuildIndex(ctx, config_obj, hunt_id) { - err := self.syncFlowTables(ctx, config_obj, hunt_id) - if err != nil { - close(output_chan) - return output_chan, 0, err - } - } - - launcher, err := services.GetLauncher(config_obj) + err := self.syncFlowTables(ctx, config_obj, hunt_id) if err != nil { close(output_chan) return output_chan, 0, err @@ -266,12 +192,17 @@ func (self HuntDispatcher) GetFlows( return output_chan, 0, err } + launcher, err := services.GetLauncher(config_obj) + if err != nil { + close(output_chan) + rs_reader.Close() + return output_chan, 0, err + } + go func() { defer close(output_chan) defer rs_reader.Close() - defer cvelo_services.Summarize("HuntDispatcher: GetFlows")() - for row := range rs_reader.Rows(ctx) { client_id, pres := row.GetString("ClientId") if !pres { @@ -280,6 +211,7 @@ func (self HuntDispatcher) GetFlows( continue } } + flow_id, pres := row.GetString("FlowId") if !pres { flow_id, pres = row.GetString("flow_id") @@ -289,6 +221,7 @@ func (self HuntDispatcher) GetFlows( } var collection_context *api_proto.FlowDetails + if options.BasicInformation { collection_context = &api_proto.FlowDetails{ Context: &flows_proto.ArtifactCollectorContext{ @@ -296,6 +229,7 @@ func (self HuntDispatcher) GetFlows( SessionId: flow_id, }, } + // If the user wants detailed flow information we need // to fetch this now. For many uses this is not // necessary so we can get away with very basic diff --git a/services/hunt_dispatcher/index.go b/services/hunt_dispatcher/index.go index aef2b5e..2be35bd 100644 --- a/services/hunt_dispatcher/index.go +++ b/services/hunt_dispatcher/index.go @@ -31,9 +31,6 @@ type HuntIndexEntry struct { func (self *HuntStorageManagerImpl) FlushIndex( ctx context.Context) error { - // Flush the index for the hunts table. - cvelo_services.Count("HuntsFlushIndex") - start_row := 0 length := 1000 diff --git a/services/indexing/indexing.go b/services/indexing/indexing.go index 680a8b1..d43e040 100644 --- a/services/indexing/indexing.go +++ b/services/indexing/indexing.go @@ -4,11 +4,8 @@ import ( "context" "errors" "sync" - "time" - "github.com/Velocidex/ttlcache/v2" cvelo_api "www.velocidex.com/golang/cloudvelo/schema/api" - "www.velocidex.com/golang/cloudvelo/services" cvelo_services "www.velocidex.com/golang/cloudvelo/services" api_proto "www.velocidex.com/golang/velociraptor/api/proto" config_proto "www.velocidex.com/golang/velociraptor/config/proto" @@ -27,9 +24,6 @@ type ElasticIndexRecord struct { type Indexer struct { config_obj *config_proto.Config ctx context.Context - - // client_id: *cvelo_api.ClientRecord - lru *ttlcache.Cache } func (self Indexer) SetIndex(client_id, term string) error { @@ -125,7 +119,8 @@ func (self Indexer) SearchIndexWithPrefix( case "label": terms := json.Format(fieldSearchQuery, "labels", term) - query := json.Format(searchlabel, terms) + query := json.Format( + searchlabel, terms) self.getIndexRecords(ctx, config_obj, query, output_chan) return @@ -164,19 +159,6 @@ func (self Indexer) FastGetApiClient( config_obj *config_proto.Config, client_id string) (*api_proto.ApiClient, error) { - // First check if the client record is actually cached. - record_any, err := self.lru.Get(client_id) - if err == nil { - record, ok := record_any.(*cvelo_api.ClientRecord) - if ok { - services.Count("FastGetApiClient (Cached)") - return _makeApiClient(record), nil - } - } - - services.Count("FastGetApiClient") - - // Nope - get it the old way. records, err := cvelo_api.GetMultipleClients( ctx, config_obj, []string{client_id}) if err != nil { @@ -187,8 +169,6 @@ func (self Indexer) FastGetApiClient( return nil, utils.NotFoundError } - self.lru.Set(client_id, records[0]) - return _makeApiClient(records[0]), nil } @@ -220,22 +200,9 @@ func _makeApiClient(client_info *cvelo_api.ClientRecord) *api_proto.ApiClient { func NewIndexingService(ctx context.Context, wg *sync.WaitGroup, config_obj *config_proto.Config) (*Indexer, error) { - - lru := ttlcache.NewCache() - lru.SetCacheSizeLimit(20000) - lru.SkipTTLExtensionOnHit(true) - lru.SetTTL(5 * time.Minute) - - go func() { - <-ctx.Done() - lru.Close() - }() - indexer := &Indexer{ config_obj: config_obj, ctx: ctx, - lru: lru, } - return indexer, nil } diff --git a/services/indexing/search_chan.go b/services/indexing/search_chan.go index 8dced64..92532dd 100644 --- a/services/indexing/search_chan.go +++ b/services/indexing/search_chan.go @@ -67,8 +67,6 @@ func (self *Indexer) searchWithTermsChan( config_obj *config_proto.Config, terms []string) (chan *api.ClientRecord, error) { - cvelo_services.Count("Indexer: searchWithTermsChan") - output_chan := make(chan *api.ClientRecord) go func() { @@ -77,10 +75,8 @@ func (self *Indexer) searchWithTermsChan( query := json.Format(strings.TrimSpace(getAllClientsQuery), strings.Join(terms, ","), "") - page_size := 1000 - // Page the query in parts. First part specifies the size. - part_query := json.Format(`{"size":%q,`, page_size) + query[1:] + part_query := `{"size":1000,` + query[1:] hits, _, err := cvelo_services.QueryElasticIds( ctx, config_obj.OrgId, "persisted", part_query) if err != nil { @@ -112,11 +108,6 @@ func (self *Indexer) searchWithTermsChan( } } - // Last page is part page. - if len(hits) < page_size { - break - } - // Get the next batch hits, _, err = cvelo_services.QueryElasticIds( ctx, config_obj.OrgId, "persisted", @@ -154,7 +145,6 @@ func (self *Indexer) SearchClientsChan( defer close(output_chan) for c := range rows { - self.lru.Set(c.ClientId, c) output_chan <- _makeApiClient(c) } }() diff --git a/services/indexing/utils.go b/services/indexing/utils.go deleted file mode 100644 index f253bea..0000000 --- a/services/indexing/utils.go +++ /dev/null @@ -1,31 +0,0 @@ -package indexing - -import ( - "context" - - config_proto "www.velocidex.com/golang/velociraptor/config/proto" - "www.velocidex.com/golang/velociraptor/services" - "www.velocidex.com/golang/velociraptor/utils" -) - -// Force all the client info to be loaded into the memory cache. -func PopulateClientInfoCache( - ctx context.Context, - config_obj *config_proto.Config) error { - - indexer, err := services.GetIndexer(config_obj) - if err != nil { - return err - } - - output_chan, err := indexer.SearchClientsChan(ctx, - nil, config_obj, "all", - utils.GetSuperuserName(config_obj)) - if err != nil { - return err - } - - for _ = range output_chan { - } - return nil -} diff --git a/services/instrument.go b/services/instrument.go index 11e928c..e9303e3 100644 --- a/services/instrument.go +++ b/services/instrument.go @@ -16,30 +16,8 @@ var ( }, []string{"operation"}, ) - - OpensearchSummary = promauto.NewSummaryVec( - prometheus.SummaryOpts{ - Name: "opensearch_operations", - Help: "Latency to access datastore.", - }, - []string{"operation"}, - ) - - // Watch operations in real time using: - // watch 'curl -s http://localhost:8003/metrics | grep -E "operations{|opensearch_latency_bucket.+Inf"' - OperationCounter = promauto.NewCounterVec( - prometheus.CounterOpts{ - Name: "operations", - Help: "Count of operations.", - }, - []string{"operation"}, - ) ) -func Count(operation string) { - OperationCounter.WithLabelValues(operation).Inc() -} - func Instrument(operation string) func() time.Duration { timer := prometheus.NewTimer(prometheus.ObserverFunc(func(v float64) { OpensearchHistorgram.WithLabelValues(operation).Observe(v) @@ -47,11 +25,3 @@ func Instrument(operation string) func() time.Duration { return timer.ObserveDuration } - -func Summarize(operation string) func() time.Duration { - timer := prometheus.NewTimer(prometheus.ObserverFunc(func(v float64) { - OpensearchSummary.WithLabelValues(operation).Observe(v) - })) - - return timer.ObserveDuration -} diff --git a/services/launcher.go b/services/launcher.go index d71aa24..63f52e3 100644 --- a/services/launcher.go +++ b/services/launcher.go @@ -5,7 +5,6 @@ import ( config_proto "www.velocidex.com/golang/velociraptor/config/proto" flows_proto "www.velocidex.com/golang/velociraptor/flows/proto" - "www.velocidex.com/golang/velociraptor/services" ) // A more efficient launcher @@ -16,14 +15,3 @@ type MultiLauncher interface { request *flows_proto.ArtifactCollectorArgs, client_ids []string) error } - -type Flusher interface { - Flush() -} - -func Flush(launcher services.Launcher) { - l, ok := launcher.Storage().(Flusher) - if ok { - l.Flush() - } -} diff --git a/services/launcher/flows.go b/services/launcher/flows.go index 04ffc4e..fc040ce 100644 --- a/services/launcher/flows.go +++ b/services/launcher/flows.go @@ -16,6 +16,25 @@ var ( const ( prefixQuery = `{"prefix": {"id": "%v"}}` regexQuery = `{"regexp": {"id": "%v[_task|_stats|_stats_completed|_completed]*"}}` + + getFlowsQuery = ` +{ + "sort": [{ + "session_id": {"order": "desc"} + }], + "query": { + "bool": { + "must": [ + {"match": {"client_id" : %q}}, + {"match": {"type": "main"}}, + {"match": {"doc_type": "collection"}} + ]} + }, + "_source": true, + "from": %q, + "size": %q +} +` ) // Are any queries currenrly running. diff --git a/services/launcher/index.go b/services/launcher/index.go index c3e2c6b..7303901 100644 --- a/services/launcher/index.go +++ b/services/launcher/index.go @@ -2,8 +2,8 @@ package launcher import ( "context" + "fmt" "sort" - "time" "github.com/Velocidex/ordereddict" cvelo_schema_api "www.velocidex.com/golang/cloudvelo/schema/api" @@ -14,7 +14,6 @@ import ( "www.velocidex.com/golang/velociraptor/json" "www.velocidex.com/golang/velociraptor/paths" "www.velocidex.com/golang/velociraptor/result_sets" - "www.velocidex.com/golang/velociraptor/services/launcher" "www.velocidex.com/golang/velociraptor/utils" ) @@ -31,10 +30,15 @@ const ( "match": { "client_id": %q } + }, { + "match": { + "type": "main" + } } ] } - } + }, + "size": 10000 } ` ) @@ -44,25 +48,7 @@ func (self *FlowStorageManager) WriteFlowIndex( config_obj *config_proto.Config, flow *flows_proto.ArtifactCollectorContext) error { - // The index of flows in the GUI. - cvelo_services.Count("WriteFlowIndex") - return self.buildIndex(ctx, config_obj, flow.ClientId) - - client_path_manager := paths.NewClientPathManager(flow.ClientId) - file_store_factory := file_store.GetFileStore(config_obj) - - // Try to open the result set. - rs_reader, err := result_sets.NewResultSetReader( - file_store_factory, - client_path_manager.FlowIndex()) - - if err != nil || self.shouldRebuildIndex( - ctx, config_obj, flow.ClientId, rs_reader) { - return self.buildIndex(ctx, config_obj, flow.ClientId) - } - - return nil } func (self *FlowStorageManager) buildIndex( @@ -70,52 +56,36 @@ func (self *FlowStorageManager) buildIndex( config_obj *config_proto.Config, client_id string) error { - cvelo_services.Count("FlowStorageManager: buildIndex") - - // Do not allow the index rebuild to be cancelled or we will end - // up with a broken index. - ctx, cancel := context.WithTimeout(context.Background(), time.Minute) - defer cancel() - - hit_chan, err := cvelo_services.QueryChan(ctx, - config_obj, 1000, config_obj.OrgId, cvelo_services.TRANSIENT, - json.Format(getCollectionsQuery, client_id), "timestamp") + query := fmt.Sprintf(getCollectionsQuery, client_id) + records, _, err := cvelo_services.QueryElasticRaw(ctx, + config_obj.OrgId, "transient", query) if err != nil { return err } - seen := make(map[string]*flows_proto.ArtifactCollectorContext) + seen := make(map[string]bool) - for hit := range hit_chan { - record := &cvelo_schema_api.ArtifactCollectorRecord{} - err = json.Unmarshal(hit, record) + var flows []*flows_proto.ArtifactCollectorContext + for _, record := range records { + item := &cvelo_schema_api.ArtifactCollectorRecord{} + err = json.Unmarshal(record, &item) if err != nil { continue } - item, err := record.ToProto() - if err != nil { + _, pres := seen[item.SessionId] + if pres { continue } - existing_record, pres := seen[item.SessionId] - if !pres { - existing_record = &flows_proto.ArtifactCollectorContext{ - ClientId: record.ClientId, - SessionId: record.SessionId, - State: item.State, - TotalRequests: item.TotalRequests, - OutstandingRequests: item.OutstandingRequests, - } - } + seen[item.SessionId] = true - seen[item.SessionId] = mergeRecords(existing_record, item) - } + flow_context, err := item.ToProto() + if err != nil { + continue + } - var flows []*flows_proto.ArtifactCollectorContext - for _, v := range seen { - launcher.UpdateFlowStats(v) - flows = append(flows, v) + flows = append(flows, flow_context) } sort.Slice(flows, func(i, j int) bool { @@ -137,93 +107,14 @@ func (self *FlowStorageManager) buildIndex( defer rs_writer.Close() for _, flow := range flows { - creator := "" - var artifacts []string - if flow.Request != nil { - artifacts = flow.Request.Artifacts - creator = flow.Request.Creator - - } else { - artifacts = flow.ArtifactsWithResults - } - summary := ordereddict.NewDict(). Set("FlowId", flow.SessionId). - Set("Artifacts", artifacts). + Set("Artifacts", flow.Request.Artifacts). Set("Created", flow.StartTime). - Set("Creator", creator). - Set("_Flow", flow) + Set("Creator", flow.Request.Creator) rs_writer.Write(summary) } return nil } - -const getLatestFlowRecord = ` -{ - "sort": [ - {"timestamp": {"order": "desc"}} - ], - "query": { - "bool": { - "must": [ - {"match": {"client_id" : %q}}, - {"match": {"doc_type" : "collection"}} - ]} - }, - "size": 1 -} -` - -// We only need to rebuild the index if the latest flow document is -// newer than the index. -func (self *FlowStorageManager) shouldRebuildIndex( - ctx context.Context, config_obj *config_proto.Config, - client_id string, - - // The rs reader of the index. - rs_reader result_sets.ResultSetReader) bool { - - // Total rows is -1 if the index was aborted. - if rs_reader == nil || rs_reader.TotalRows() <= 0 { - return true - } - - // Within 1 minute we do not rebuild the index - we are ok with - // the index being 1 minute out. - now := utils.GetTime().Now() - - min_flow_cache_time_min := time.Minute - if self.cloud_config != nil && - self.cloud_config.MinFlowCacheTimeMin != 0 { - min_flow_cache_time_min = time.Minute * - time.Duration(self.cloud_config.MinFlowCacheTimeMin) - } - - if now.Sub(rs_reader.MTime()) < min_flow_cache_time_min { - return false - } - - hit, err := cvelo_services.GetElasticRecordByQuery(ctx, - config_obj.OrgId, cvelo_services.TRANSIENT, json.Format( - getLatestFlowRecord, client_id)) - - // If there are no flows at all in this client, we dont need to - // build any indexes. - if err != nil || len(hit) == 0 { - return false - } - - item := &cvelo_schema_api.ArtifactCollectorRecord{} - err = json.Unmarshal(hit, item) - if err == nil { - last_modified := time.Unix(0, item.Timestamp) - if rs_reader.MTime().After(last_modified) { - // Skip the update if the result set is newer than the - // last_modified record. - return false - } - } - return true -} diff --git a/services/launcher/launcher.go b/services/launcher/launcher.go index 702c908..023a6f2 100644 --- a/services/launcher/launcher.go +++ b/services/launcher/launcher.go @@ -3,10 +3,7 @@ package launcher import ( "context" "sync" - "time" - "github.com/Velocidex/ttlcache/v2" - "www.velocidex.com/golang/cloudvelo/config" config_proto "www.velocidex.com/golang/velociraptor/config/proto" "www.velocidex.com/golang/velociraptor/services" "www.velocidex.com/golang/velociraptor/services/launcher" @@ -20,24 +17,10 @@ type Launcher struct { func NewLauncherService( ctx context.Context, wg *sync.WaitGroup, - config_obj *config_proto.Config, - cloud_config *config.ElasticConfiguration) (services.Launcher, error) { - - lru := ttlcache.NewCache() - lru.SetCacheSizeLimit(20000) - lru.SkipTTLExtensionOnHit(true) - lru.SetTTL(5 * time.Minute) - - go func() { - <-ctx.Done() - lru.Close() - }() + config_obj *config_proto.Config) (services.Launcher, error) { launcher_service := &launcher.Launcher{ - Storage_: &FlowStorageManager{ - cache: lru, - cloud_config: cloud_config, - }, + Storage_: &FlowStorageManager{}, } return &Launcher{ diff --git a/services/launcher/storage.go b/services/launcher/storage.go index 694fc28..3dd11a1 100644 --- a/services/launcher/storage.go +++ b/services/launcher/storage.go @@ -5,10 +5,7 @@ import ( "errors" "fmt" "io" - "sync" - "github.com/Velocidex/ttlcache/v2" - "www.velocidex.com/golang/cloudvelo/config" "www.velocidex.com/golang/cloudvelo/schema/api" cvelo_schema_api "www.velocidex.com/golang/cloudvelo/schema/api" cvelo_services "www.velocidex.com/golang/cloudvelo/services" @@ -25,61 +22,8 @@ import ( "www.velocidex.com/golang/velociraptor/utils" ) -// Written in the flow index -type FlowSummary struct { - FlowId string `json:"FlowId"` - Artifacts []string `json:"Artifacts"` - Created uint64 `json:"Created"` - Creator string `json:"Creator"` - Flow *flows_proto.ArtifactCollectorContext `json:"_Flow"` -} - -func (self *FlowSummary) Summary() *services.FlowSummary { - return &services.FlowSummary{ - FlowId: self.FlowId, - Artifacts: self.Artifacts, - Created: self.Created, - Creator: self.Creator, - } -} - -type FlowCacheItem struct { - mu sync.Mutex - data map[string]*flows_proto.ArtifactCollectorContext -} - -func (self *FlowCacheItem) Set(key string, value *flows_proto.ArtifactCollectorContext) { - self.mu.Lock() - defer self.mu.Unlock() - - self.data[key] = value -} - -func (self *FlowCacheItem) Get(key string) (*flows_proto.ArtifactCollectorContext, bool) { - self.mu.Lock() - defer self.mu.Unlock() - - res, pres := self.data[key] - return res, pres -} - -func NewFlowCacheItem() *FlowCacheItem { - return &FlowCacheItem{ - data: make(map[string]*flows_proto.ArtifactCollectorContext), - } -} - type FlowStorageManager struct { launcher.FlowStorageManager - - // client_id -> FlowCacheItem : map[flow_id]*flows_proto.ArtifactCollectorContext - cache *ttlcache.Cache - - cloud_config *config.ElasticConfiguration -} - -func (self *FlowStorageManager) Flush() { - self.cache.Purge() } func (self *FlowStorageManager) WriteFlow( @@ -139,58 +83,18 @@ func (self *FlowStorageManager) ListFlows( options result_sets.ResultSetOptions, offset int64, length int64) ([]*services.FlowSummary, int64, error) { - cvelo_services.Count("ListFlows") - - flows, total, err := self._ListFlows(ctx, config_obj, - client_id, options, offset, length) + err := self.buildIndex(ctx, config_obj, client_id) if err != nil { return nil, 0, err } - res := make([]*services.FlowSummary, 0, len(flows)) - cache_item := NewFlowCacheItem() - for _, i := range flows { - res = append(res, i.Summary()) - cache_item.Set(i.FlowId, i.Flow) - } - self.cache.Set(client_id, cache_item) - - return res, total, nil -} - -func (self *FlowStorageManager) _ListFlows( - ctx context.Context, - config_obj *config_proto.Config, - client_id string, - options result_sets.ResultSetOptions, - offset int64, length int64) ([]*FlowSummary, int64, error) { - - result := []*FlowSummary{} + result := []*services.FlowSummary{} client_path_manager := paths.NewClientPathManager(client_id) file_store_factory := file_store.GetFileStore(config_obj) - - // Try to open the result set. rs_reader, err := result_sets.NewResultSetReaderWithOptions( ctx, config_obj, file_store_factory, client_path_manager.FlowIndex(), options) - - // Index does not exist yet. - if err != nil || self.shouldRebuildIndex( - ctx, config_obj, client_id, rs_reader) { - - // Force a rebuild of the index. - err1 := self.buildIndex(ctx, config_obj, client_id) - if err1 != nil { - return nil, 0, fmt.Errorf("buildIndex: %w", err) - } - - // Try to open it again. Hopefully it works now. - rs_reader, err = result_sets.NewResultSetReaderWithOptions( - ctx, config_obj, file_store_factory, - client_path_manager.FlowIndex(), options) - } - - if err != nil || rs_reader == nil || rs_reader.TotalRows() <= 0 { + if err != nil || rs_reader.TotalRows() <= 0 { return result, 0, nil } @@ -210,7 +114,7 @@ func (self *FlowStorageManager) _ListFlows( } for serialized := range json_chan { - summary := &FlowSummary{} + summary := &services.FlowSummary{} err = json.Unmarshal(serialized, summary) if err == nil { result = append(result, summary) @@ -263,46 +167,6 @@ func (self *FlowStorageManager) LoadCollectionContext( return &flows_proto.ArtifactCollectorContext{}, nil } - // Hit the cache for the collection context - flows_any, err := self.cache.Get(client_id) - if err != nil { - cvelo_services.Count("LoadCollectionContext") - - // This should populate the cache if needed. - _, _, err := self.ListFlows(ctx, config_obj, - client_id, result_sets.ResultSetOptions{}, 0, 10000) - if err != nil { - return nil, err - } - - // Try to get it again from the cache. Should be there this - // time. - flows_any, err = self.cache.Get(client_id) - - } else { - cvelo_services.Count("LoadCollectionContext (Cached)") - } - - flows, ok := flows_any.(*FlowCacheItem) - if ok { - hit, ok := flows.Get(flow_id) - if ok { - return hit, nil - } - } - - return nil, utils.NotFoundError -} - -// The old slow version of LoadCollectionContext. -// TODO: Remove when the new code is working well. -func (self *FlowStorageManager) LoadCollectionContextSlow( - ctx context.Context, - config_obj *config_proto.Config, - client_id, flow_id string) (*flows_proto.ArtifactCollectorContext, error) { - - cvelo_services.Count("LoadCollectionContext") - hit_chan, err := cvelo_services.QueryChan(ctx, config_obj, 1000, config_obj.OrgId, "transient", json.Format(getFlowDetailsQuery, client_id, flow_id), "timestamp") diff --git a/services/notebook/timeline_storer.go b/services/notebook/timeline_storer.go index 3610431..3159112 100644 --- a/services/notebook/timeline_storer.go +++ b/services/notebook/timeline_storer.go @@ -51,6 +51,23 @@ const ( "size": %q, "from": %q } +` + + query_for_specific_supertimeline = ` +{ + "sort": [ + { + "timestamp": {"order": "desc"} + }], + "query": { + "bool": { + "must": [ + {"match": {"notebook_id": %q}}, + {"match": {"supertimeline_name": %q}}, + {"match": {"type": "Supertimeline"}} + ]} + } +} ` ) @@ -120,8 +137,6 @@ func (self SuperTimelineStorer) List(ctx context.Context, count := 1000 offset := 0 - cvelo_services.Count("SuperTimelineStorer: List") - query := json.Format(query_for_supertimelines, notebook_id, count, offset) hits, _, err := cvelo_services.QueryElasticRaw( ctx, self.config_obj.OrgId, "persisted", query) diff --git a/services/orgs/lazy.go b/services/orgs/lazy.go index 2cb2844..9ec8572 100644 --- a/services/orgs/lazy.go +++ b/services/orgs/lazy.go @@ -46,12 +46,6 @@ type LazyServiceContainer struct { broadcast services.BroadcastService journal services.JournalService - - client_info services.ClientInfoManager - - launcher services.Launcher - - indexer services.Indexer } func (self *LazyServiceContainer) FrontendManager() (services.FrontendManager, error) { @@ -94,38 +88,16 @@ func (self *LazyServiceContainer) NotebookManager() (services.NotebookManager, e return notebook.NewNotebookManagerService(self.ctx, self.wg, self.config_obj), nil } -func (self *LazyServiceContainer) Launcher() (res services.Launcher, err error) { - self.mu.Lock() - defer self.mu.Unlock() - - if self.launcher == nil { - self.launcher, err = launcher.NewLauncherService( - self.ctx, self.wg, self.config_obj, - self.cloud_config) - if err != nil { - return nil, err - } - } - - return self.launcher, nil +func (self *LazyServiceContainer) Launcher() (services.Launcher, error) { + return launcher.NewLauncherService(self.ctx, self.wg, self.config_obj) } func (self *LazyServiceContainer) HuntDispatcher() (services.IHuntDispatcher, error) { return hunt_dispatcher.NewHuntDispatcher(self.ctx, self.wg, self.config_obj) } -func (self *LazyServiceContainer) Indexer() (res services.Indexer, err error) { - self.mu.Lock() - defer self.mu.Unlock() - - if self.indexer == nil { - self.indexer, err = indexing.NewIndexingService(self.ctx, self.wg, self.config_obj) - if err != nil { - return nil, err - } - } - - return self.indexer, nil +func (self *LazyServiceContainer) Indexer() (services.Indexer, error) { + return indexing.NewIndexingService(self.ctx, self.wg, self.config_obj) } func (self *LazyServiceContainer) Scheduler() (services.Scheduler, error) { @@ -161,19 +133,8 @@ func (self *LazyServiceContainer) Journal() (res services.JournalService, err er return self.journal, nil } -func (self *LazyServiceContainer) ClientInfoManager() (res services.ClientInfoManager, err error) { - self.mu.Lock() - defer self.mu.Unlock() - - if self.client_info == nil { - self.client_info, err = client_info.NewClientInfoManager( - self.config_obj) - if err != nil { - return nil, err - } - } - - return self.client_info, nil +func (self *LazyServiceContainer) ClientInfoManager() (services.ClientInfoManager, error) { + return client_info.NewClientInfoManager(self.config_obj) } func (self *LazyServiceContainer) Inventory() (services.Inventory, error) { diff --git a/services/scheduler/scheduler.go b/services/scheduler/scheduler.go index 57a1367..7bdc4b7 100644 --- a/services/scheduler/scheduler.go +++ b/services/scheduler/scheduler.go @@ -77,8 +77,6 @@ type ElasticScheduler struct{} func (self *ElasticScheduler) getOneJob(ctx context.Context, worker_id, queue string) (*services.SchedulerJob, error) { - cvelo_services.Count("ElasticScheduler: getOneJob") - // Schedulers are global across all orgs. org_id := services.ROOT_ORG_ID diff --git a/services/users/storage.go b/services/users/storage.go index b29b5ab..b0a18bb 100644 --- a/services/users/storage.go +++ b/services/users/storage.go @@ -180,8 +180,6 @@ func (self *UserStorageManager) ListAllUsers( func (self *UserStorageManager) SetUserOptions(ctx context.Context, username string, options *api_proto.SetGUIOptionsRequest) error { - cvelo_services.Count("SetUserOptions") - // Merge the old options with the new options old_options, err := self.GetUserOptions(ctx, username) if err != nil { @@ -253,8 +251,6 @@ const doc_type_query = ` func (self *UserStorageManager) GetUserOptions(ctx context.Context, username string) ( *api_proto.SetGUIOptionsRequest, error) { - cvelo_services.Count("GetUserOptions") - serialized, err := cvelo_services.GetElasticRecordByQuery(ctx, services.ROOT_ORG_ID, "persisted", json.Format(doc_type_query, username+"_options", "user_options")) diff --git a/testsuite/server_config.go b/testsuite/server_config.go index c654aa2..bc2b80c 100644 --- a/testsuite/server_config.go +++ b/testsuite/server_config.go @@ -282,7 +282,6 @@ Cloud: credentials_secret: password no_verify_cert: true foreman_interval_seconds: 5 - min_flow_cache_time_min: -1 ` const writeback_file = `