x-pack/filebeat/input/cel: fix handling of non-200/non-429 status codes

CHANGELOG.next.asciidoc

@@ -62,6 +62,7 @@ https://github.com/elastic/beats/compare/v8.2.0\...main[Check the HEAD diff]
 - Fix input cancellation handling when HTTP client does not support contexts. {issue}33962[33962] {pull}33968[33968]
 - Update mito CEL extension library to v0.0.0-20221207004749-2f0f2875e464 {pull}33974[33974]
 - Fix CEL result deserialisation when evaluation fails. {issue}33992[33992] {pull}33996[33996]
+- Fix handling of non-200/non-429 status codes. {issue}33999[33999] {pull}34002[34002]
 
 *Heartbeat*
 - Fix broken zip URL monitors. NOTE: Zip URL Monitors will be removed in version 8.7 and replaced with project monitors. {pull}33723[33723]

x-pack/filebeat/input/cel/input.go

@@ -529,10 +529,14 @@ func handleResponse(log *logp.Logger, state map[string]interface{}, limiter *rat
 					waitUntil = t
 				}
 			}
-			fallthrough
-		default:
-			delete(state, "events")
 			return false, waitUntil, nil
+		default:
+			status := http.StatusText(statusCode)
+			if status == "" {
+				status = "unknown status code"
+			}
+			state["events"] = errorMessage(fmt.Sprintf("failed http request with %s: %d", status, statusCode))
+			return true, time.Time{}, nil
 		}
 	}
 	return true, waitUntil, nil
@@ -849,29 +853,33 @@ func evalWith(ctx context.Context, prg cel.Program, state map[string]interface{}
 		err = e
 	}
 	if err != nil {
-		state["events"] = map[string]interface{}{"error.message": fmt.Sprintf("failed eval: %v", err)}
+		state["events"] = errorMessage(fmt.Sprintf("failed eval: %v", err))
 		return state, fmt.Errorf("failed eval: %w", err)
 	}
 
 	v, err := out.ConvertToNative(reflect.TypeOf(&structpb.Value{}))
 	if err != nil {
-		state["events"] = map[string]interface{}{"error.message": fmt.Sprintf("failed proto conversion: %v", err)}
+		state["events"] = errorMessage(fmt.Sprintf("failed proto conversion: %v", err))
 		return state, fmt.Errorf("failed proto conversion: %w", err)
 	}
 	b, err := protojson.MarshalOptions{Indent: ""}.Marshal(v.(proto.Message))
 	if err != nil {
-		state["events"] = map[string]interface{}{"error.message": fmt.Sprintf("failed native conversion: %v", err)}
+		state["events"] = errorMessage(fmt.Sprintf("failed native conversion: %v", err))
 		return state, fmt.Errorf("failed native conversion: %w", err)
 	}
 	var res map[string]interface{}
 	err = json.Unmarshal(b, &res)
 	if err != nil {
-		state["events"] = map[string]interface{}{"error.message": fmt.Sprintf("failed json conversion: %v", err)}
+		state["events"] = errorMessage(fmt.Sprintf("failed json conversion: %v", err))
 		return state, fmt.Errorf("failed json conversion: %w", err)
 	}
 	return res, nil
 }
 
+func errorMessage(msg string) map[string]interface{} {
+	return map[string]interface{}{"error": map[string]interface{}{"message": msg}}
+}
+
 // retryLog is a shim for the retryablehttp.Client.Logger.
 type retryLog struct{ log *logp.Logger }
 

x-pack/filebeat/input/cel/input_test.go

@@ -985,7 +985,9 @@ bytes(get(state.url).Body).decode_json().records.map(r,
 		handler: defaultHandler(http.MethodGet, ""),
 		want: []map[string]interface{}{
 			{
-				"error.message": "failed eval: no such overload", // This is the best we get for some errors from CEL.
+				"error": map[string]interface{}{
+					"message": "failed eval: no such overload", // This is the best we get for some errors from CEL.
+				},
 			},
 		},
 	},