Jsonpb parse behavior changed breaking semantics (Velocidex#988)

Unmarshal of empty string returns an error now so we need to special
case it.

artifacts/definitions/Splunk/Flows/Upload.yaml

@@ -19,7 +19,7 @@ description: |
        * Go to Settings > Data Inputs > HTTP Event Collector > Global Settings
        * Ensure `All Tokens` is set to ENABLED
        * Copy the HTTP Port Number for this event
-       
+
        > Note: `Enable SSL` only works if SSL is properly configured on your
        Splunk server -- meaning you have proper certificates and DNS. If you are
        accessing your Splunk instance by IP, `Enable SSL` should be set to OFF.
@@ -46,14 +46,14 @@ parameters:
      default: false
      description: |
       SSL configured with the event collector. This is false by default.
-     
+
 sources:
-  - query: |  
+  - query: |
         LET completions = SELECT * FROM watch_monitoring(
                      artifact="System.Flow.Completion")
-                 WHERE Flow.artifacts_with_results =~ ArtifactNameRegex 
+                 WHERE Flow.artifacts_with_results =~ ArtifactNameRegex
                      AND log(message=Flow.artifacts_with_results)
-        
+
         LET documents = SELECT * FROM foreach(row=completions,
                   query={
                      SELECT * FROM foreach(
@@ -70,11 +70,11 @@ sources:
                                 artifact=_value)
                          })
                   })
-              
+
         SELECT * FROM splunk_upload(
         query = documents,
         url = url,
         token = token,
-        index = velociraptor,
+        index = index,
         skip_verify = SSL
         )

datastore/datastore_test.go

@@ -36,6 +36,11 @@ func (self BaseTestSuite) TestSetGetSubject() {
 	err = self.datastore.GetSubject(self.config_obj, urn+"foo", read_message)
 	assert.NoError(self.T(), err)
 
+	// Same for json files.
+	err = self.datastore.GetSubject(
+		self.config_obj, urn+"foo.json", read_message)
+	assert.NoError(self.T(), err)
+
 	// Delete the subject
 	err = self.datastore.DeleteSubject(self.config_obj, urn)
 	assert.NoError(self.T(), err)

datastore/filebased.go

@@ -148,6 +148,14 @@ func (self *FileBaseDataStore) GetSubject(
 		return err
 	}
 
+	// Clear the target.
+	proto.Reset(message)
+
+	// Nothing to do - no data.
+	if len(serialized_content) == 0 {
+		return nil
+	}
+
 	if strings.HasSuffix(urn, ".json") {
 		return protojson.Unmarshal(serialized_content, message)
 	}

server/server.go

@@ -99,6 +99,7 @@ func (self *Server) adjustConcurrency(
 		}
 
 		// Adjust concurrency
+		self.logger.Debug("Adjusting concurrency from %v to %v", concurrency, new_concurrency)
 		concurrency = new_concurrency
 
 		// We are using up less memory than
@@ -113,6 +114,7 @@ func (self *Server) adjustConcurrency(
 		}
 
 		// Adjust concurrency
+		self.logger.Debug("Adjusting concurrency from %v to %v", concurrency, concurrency+delta)
 		concurrency += delta
 
 	} else {
@@ -122,6 +124,7 @@ func (self *Server) adjustConcurrency(
 	}
 
 	// Install the new concurrency controller.
+
 	targetConcurrency.Set(float64(concurrency))
 	heapSize.Set(float64(s.Alloc))
 	self.mu.Lock()
@@ -137,7 +140,6 @@ func (self *Server) ManageConcurrency(max_concurrency uint64, target_heap_size u
 
 	for {
 		new_concurrency := self.adjustConcurrency(max_concurrency, target_heap_size, concurrency)
-		self.logger.Debug("Adjusting concurrency from %v to %v", concurrency, new_concurrency)
 		concurrency = new_concurrency
 
 		// Wait for a minute and check again.