Merge branch 'versions/0.10' into dev

CHANGES.txt

@@ -48,6 +48,8 @@ Features
 
 * Added `separator` and `maximum_depth` settings to JSON decoder (#1809).
 
+* Added `status_field` setting to http_status filter (#1876)
+
 Bug Handling
 ------------
 
@@ -57,7 +59,7 @@ Bug Handling
 * Don't use `os.Exit` in the pipeline's main `Run` function or else any
   wrapping deffered functions (such as those that output the cpu and mem
   profiles) won't get called.
-  
+
 * Stop zero-length records with no error ending the splitter if there's
   more data to read (#1561)
 
@@ -67,6 +69,11 @@ Bug Handling
   names by always triggering LogstreamSet rescan before checking location in
   stream (#1452).
 
+* JSON decoder no longer exits when flattening the field structure fails.
+
+* Fixed bug where HttpOutput would discard the response body before trying to
+  use the response body when response returned an error status code.
+
 0.10.0 (2015-12-30)
 ===================
 

docs/source/config/index.rst

@@ -169,7 +169,7 @@ Config:
 - log_flags (int):
     Control the prefix for STDOUT and STDERR logs. Common values are 3 (date
     and time, the default) or 0 (no prefix). See
-    `https://golang.org/pkg/log/#pkg-constants Go documentation`_ for details.
+    `Go documentation <https://golang.org/pkg/log/#pkg-constants>`_ for details.
 
 - full_buffer_max_retries (int):
     When Heka shuts down due to a buffer filling to capacity, the next time

plugins/http/http_output.go

@@ -148,15 +148,15 @@ func (o *HttpOutput) request(or pipeline.OutputRunner, outBytes []byte) (err err
 		return fmt.Errorf("Error making HTTP request: %s", err.Error())
 	}
 	defer resp.Body.Close()
-	io.Copy(ioutil.Discard, resp.Body)
-
 	if resp.StatusCode >= 400 {
 		body, err := ioutil.ReadAll(resp.Body)
 		if err != nil {
 			return fmt.Errorf("Error reading HTTP response: %s", err.Error())
 		}
 		return fmt.Errorf("HTTP Error code returned: %d %s - %s",
 			resp.StatusCode, resp.Status, string(body))
+	} else {
+		io.Copy(ioutil.Discard, resp.Body)
 	}
 	return
 }

sandbox/lua/decoders/json.lua

@@ -171,8 +171,8 @@ local msg = {
 }
 
 function process_message()
-    local pok, json = pcall(cjson.decode, read_message("Payload"))
-    if not pok then return -1, "Failed to decode JSON." end
+    local ok, json = pcall(cjson.decode, read_message("Payload"))
+    if not ok then return -1, "Failed to decode JSON." end
 
     -- keep payload, or not
     if payload_keep then
@@ -209,7 +209,10 @@ function process_message()
 
     -- flatten and assign remaining fields to heka fields
     local flat = {}
-    util.table_to_fields(json, flat, nil, separator, max_depth)
+    if not pcall(util.table_to_fields, json, flat, nil, separator, max_depth) then
+        return -1, "Failed to flatten message."
+    end
+
     msg.Fields = flat
 
     if not pcall(inject_message, msg) then

sandbox/lua/filters/http_status.lua

@@ -3,8 +3,8 @@
 -- file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
 --[[
-Graphs HTTP status codes using the numeric Fields[status] variable collected
-from web server access logs.
+Graphs HTTP status codes using the numeric status code collected from
+web server access logs.
 
 Config:
 
@@ -16,6 +16,9 @@ Config:
     Sets the size of the sliding window i.e., 1440 rows representing 60
     seconds per row is a 24 sliding hour window with 1 minute resolution.
 
+- status_field (string, optional, default "status")
+    Sets the message field containing the numeric HTTP status code.
+
 - anomaly_config (string, optional)
     See :ref:`sandbox_anomaly_module`.
 
@@ -58,6 +61,7 @@ local anomaly       = require "anomaly"
 local title             = "HTTP Status"
 local rows              = read_config("rows") or 1440
 local sec_per_row       = read_config("sec_per_row") or 60
+local status_field      = read_config("status_field") or "status"
 local anomaly_config    = anomaly.parse_config(read_config("anomaly_config"))
 local alert_throttle    = read_config("alert_throttle") or 3600
 alert.set_throttle(alert_throttle * 1e9)
@@ -74,7 +78,7 @@ local HTTP_UNKNOWN = status:set_header(6, "HTTP_UNKNOWN")
 
 function process_message ()
     local ts = read_message("Timestamp")
-    local sc = read_message("Fields[status]")
+    local sc = read_message("Fields["..status_field.."]")
     if type(sc) ~= "number" then return -1 end
 
     local col = sc/100

sandbox/plugins/sandbox_filters_test.go

@@ -583,6 +583,7 @@ func FilterSpec(c gs.Context) {
 		conf.Config = make(map[string]interface{})
 		conf.Config["rows"] = int64(2)
 		conf.Config["sec_per_row"] = int64(1)
+		conf.Config["status_field"] = "status_code"
 
 		timer := make(chan time.Time, 1)
 		errChan := make(chan error, 1)
@@ -594,7 +595,7 @@ func FilterSpec(c gs.Context) {
 			close(retPackChan)
 		}()
 
-		field, _ := message.NewField("status", 0, "")
+		field, _ := message.NewField("status_code", 0, "")
 		msg := &message.Message{}
 		msg.SetTimestamp(0)
 		msg.AddField(field)