feat(inputs.firehose): Add new plugin #15988

syedmhashim · 2024-10-07T10:44:05Z

Summary

The firehose input plugin would be used to receive data from AWS Data Firehose.

Checklist

No AI generated code was used in this PR

Related issues

resolves #15870

feat(inputs/firehose): add input plugin for AWS Data Firehose

telegraf-tiger · 2024-10-07T11:08:56Z

Download PR build artifacts for linux_amd64.tar.gz, darwin_arm64.tar.gz, and windows_amd64.zip.
Downloads for additional architectures and packages are available below.

☺️ This pull request doesn't significantly change the Telegraf binary size (less than 1%)

📦 Click here to get additional PR build artifacts

Artifact URLs

DEB	RPM	TAR GZ	ZIP
amd64.deb	aarch64.rpm	darwin_amd64.tar.gz	windows_amd64.zip
arm64.deb	armel.rpm	darwin_arm64.tar.gz	windows_arm64.zip
armel.deb	armv6hl.rpm	freebsd_amd64.tar.gz	windows_i386.zip
armhf.deb	i386.rpm	freebsd_armv7.tar.gz
i386.deb	ppc64le.rpm	freebsd_i386.tar.gz
mips.deb	riscv64.rpm	linux_amd64.tar.gz
mipsel.deb	s390x.rpm	linux_arm64.tar.gz
ppc64el.deb	x86_64.rpm	linux_armel.tar.gz
riscv64.deb		linux_armhf.tar.gz
s390x.deb		linux_i386.tar.gz
		linux_mips.tar.gz
		linux_mipsel.tar.gz
		linux_ppc64le.tar.gz
		linux_riscv64.tar.gz
		linux_s390x.tar.gz

srebhan

@syedmhashim I do have some comments, mostly on style, so overall this looks quite promising. I urge you to keep the code as simple and less nested as possible to ease review and debugging. And of course some unit-tests would be great!

srebhan · 2024-10-08T09:35:00Z

plugins/inputs/firehose/README.md

@@ -0,0 +1,123 @@
+# AWS Data Firehose HTTP Listener Input Plugin
+
+Firehose HTTP Listener is a service input plugin that listens for metrics sent


Suggested change

Firehose HTTP Listener is a service input plugin that listens for metrics sent

This is a service input plugin that listens for metrics sent

srebhan · 2024-10-08T09:37:19Z

plugins/inputs/firehose/README.md

+  data_format = "json_v2"
+  [[inputs.firehose.json_v2]]
+    timestamp_path = "timestamp"
+    timestamp_format = "unix_ms"
+    [[inputs.firehose.json_v2.tag]]
+      path = "type"
+    [[inputs.firehose.json_v2.field]]
+      path = "message"


Can we please either use the xpath parser here or use a simple string value or somethin? Let's not promote the json_v2 parser more than necessary, especially when dealing with arrays...

srebhan · 2024-10-08T09:59:35Z

plugins/inputs/firehose/firehose.go

+	for _, method := range allowedMethods {
+		if r.req.Method == method {
+			isAcceptedMethod = true
+			break
+		}
+	}
+	if !isAcceptedMethod {
+		r.responseStatusCode = http.StatusMethodNotAllowed
+		return fmt.Errorf("%s method not allowed", r.req.Method)
+	}
+
+	contentType := r.req.Header.Get("content-type")
+	if contentType != "application/json" {
+		r.responseStatusCode = http.StatusBadRequest
+		return fmt.Errorf("%s content type not accepted", contentType)
+	}
+
+	contentEncoding := r.req.Header.Get("content-encoding")
+	if contentEncoding != "" && contentEncoding != "gzip" {
+		r.responseStatusCode = http.StatusBadRequest
+		return fmt.Errorf("%s content encoding not accepted", contentEncoding)
+	}
+
+	err := r.extractBody()
+	if err != nil {
+		return err
+	}
+
+	requestID := r.req.Header.Get(requestIDHeader)
+	if requestID == "" {
+		r.responseStatusCode = http.StatusBadRequest
+		return fmt.Errorf("%s header is not set", requestIDHeader)
+	}
+
+	if requestID != r.body.RequestID {
+		r.responseStatusCode = http.StatusBadRequest
+		return fmt.Errorf("requestId in the body does not match the value of the request header, %s", requestIDHeader)
+	}
+
+	return nil
+}
+
+func (r *firehoseRequest) extractBody() error {
+	encoding := r.req.Header.Get("content-encoding")
+	switch encoding {
+	case "gzip":
+		g, err := gzip.NewReader(r.req.Body)
+		if err != nil {
+			r.responseStatusCode = http.StatusBadRequest
+			return fmt.Errorf("unable to decode body - %s", err.Error())
+		}
+		defer g.Close()
+		err = json.NewDecoder(g).Decode(&r.body)
+		if err != nil {
+			r.responseStatusCode = http.StatusBadRequest
+			return err
+		}
+	default:
+		defer r.req.Body.Close()
+		err := json.NewDecoder(r.req.Body).Decode(&r.body)
+		if err != nil {
+			r.responseStatusCode = http.StatusBadRequest
+			return err
+		}
+	}
+	return nil
+}
+
+func (r *firehoseRequest) decodeData() ([][]byte, bool) {
+	// decode base64-encoded data and return them as a slice of byte slices
+	decodedData := make([][]byte, 0)
+	for _, record := range r.body.Records {
+		data, err := base64.StdEncoding.DecodeString(record.EncodedData)
+		if err != nil {
+			return nil, false
+		}
+		decodedData = append(decodedData, data)
+	}
+	return decodedData, true
+}
+
+func (r *firehoseRequest) sendResponse(res http.ResponseWriter) error {
+	responseBody := struct {
+		RequestID    string `json:"requestId"`
+		Timestamp    int64  `json:"timestamp"`
+		ErrorMessage string `json:"errorMessage,omitempty"`
+	}{
+		RequestID:    r.req.Header.Get(requestIDHeader),
+		Timestamp:    time.Now().Unix(),
+		ErrorMessage: statusCodeToMessage[r.responseStatusCode],
+	}
+	response, err := json.Marshal(responseBody)
+	if err != nil {
+		return err
+	}
+	res.Header().Set("content-type", "application/json")
+	res.WriteHeader(r.responseStatusCode)
+	_, err = res.Write(response)
+	return err
+}


I suggest moving this to an own file to make the plugin file more readable. Is there an upstream definition of those types in the AWS SDK v2?

Good point! I'll have a look. If I'm not able to find something in the AWS SDK v2, then I'll definitely move this to a new file

done. moved to a separate file

srebhan · 2024-10-08T10:00:59Z

plugins/inputs/firehose/firehose.go

+//go:embed sample.conf
+var sampleConfig string
+
+var once sync.Once


This should be per plugin-instance and not globally.

Noted. Btw I do see this declared globally in every other (input) plugins

srebhan · 2024-10-08T10:02:13Z

plugins/inputs/firehose/firehose.go

+)
+
+// TimeFunc provides a timestamp for the metrics
+type TimeFunc func() time.Time


Why do you need this?

I saw this in http_listener_v2 as well as in influxdb_v2_listener. Thought it's used somewhere downstream in telegraf. If this is not the case, please let me know and I'll remove this

Please remove. We had this in the dark age where we could not test metrics without pinning the timestamp...

srebhan · 2024-10-08T11:21:24Z

plugins/inputs/firehose/firehose.go

+
+		parameters, ok := parameters["commonAttributes"].(map[string]interface{})
+		if !ok {
+			f.Log.Warn(formatLog(r.body.RequestID, "Invalid value for header "+commonAttributesHeader))


Well I don't think you should simply continue here rather return here...

I thought of parameter tags as something optional. So if were able to extract and find them, we'd add to the metrics otherwise we'll ignore. Thus, I'm just emitting a warning log. Now that you mentioned, it does makes sense to maybe respond with bad request status and return here. Agree?

Oh I see, but I would then at least skip the for loop below as this is known to not return anything but just burns some cycles...

srebhan · 2024-10-08T11:22:30Z

plugins/inputs/firehose/firehose.go

+		for _, parameter := range f.ParameterTags {
+			if value, ok := parameters[parameter]; ok {
+				for _, m := range metrics {
+					m.AddTag(parameter, value.(string))
+				}
+			}
+		}
+	}
+
+	for _, m := range metrics {
+		if f.PathTag {
+			m.AddTag(pathTag, req.URL.Path)
+		}
+		f.acc.AddMetric(m)
+	}


I really would collect all tags first and then iterate over the metrics once to add them.

Okay. Does the following makes sense:

if f.PathTag { for _, m := range metrics { m.AddTag(pathTag, req.URL.Path) } } for _, m := range metrics { f.acc.AddMetric(m) }

What I want is (pseudocode)

tags := make(map[string]string, len(f.ParameterTags)+1) if f.pathTag { tags[f.pathTag] = req.URL.Path } for _, parameter := range f.ParameterTags { if value, found := parameters[parameter]; found { if v, ok := value.(string); ok { tags[parameter] = v } } } for _, m := range metrics { for k, v := range tags { m.AddTag(k , v) } f.acc.AddMetric(m) }

Actually f.PathTag variable is a boolean. When set to True, we add a tag, firehose_http_path to each metric with value set to the request path, i.e, req.URL.Path. It's unrelated to f.ParameterTags. The purpose of this was to allow the plugin to receive request on more than one request paths. So with the following configuration:

[[inputs.firehose]] ## Address and port to host HTTP listener on service_address = ":8080" ## Paths to listen to. paths = ["/telegraf/dev", "/telegraf/stage"] ## Save path as firehose_http_path tag if set to true path_tag = true

our input plugin would accept requests for both of these paths. This would allow us to have a single telegraf server running, serving both our dev and stage environments. If we want, we can also add the prod environment here or have a separate telegraf server running for it. And if the f.PathTag is set to True we'll add a tag firehose_http_path to each metric with it's value set to the actual request path e.g telegraf/dev. This is something I thought could be useful and is optional therefore. I did add comments about this in the readme and sample config

srebhan · 2024-10-08T11:22:52Z

plugins/inputs/firehose/firehose.go

+func formatLog(requestID, message string) string {
+	return fmt.Sprintf(logFormat, requestID, message)
+}


Please get rid of this function!

srebhan · 2024-10-08T11:23:36Z

plugins/inputs/firehose/firehose.go

+			ServiceAddress: ":8080",
+			TimeFunc:       time.Now,
+			Paths:          []string{"/telegraf"},


Please set the default values of ServiceAddress and Paths in Init() to reuse it in tests.

Makes sense. Btw if it's only for tests can't we do it the way it's done here? Just a thought

TBH, I hate these types of indirections. Imagine a test fails and you quickly want to look at what's going on. You immediately need to jump around in code to see how a simple test case is set up... :-( So please avoid this if possible!

srebhan · 2024-10-08T11:23:46Z

plugins/inputs/firehose/firehose.go

+			ServiceAddress: ":8080",
+			TimeFunc:       time.Now,
+			Paths:          []string{"/telegraf"},
+			close:          make(chan struct{}),


Also do in Init().

syedmhashim · 2024-10-09T06:43:40Z

@syedmhashim I do have some comments, mostly on style, so overall this looks quite promising. I urge you to keep the code as simple and less nested as possible to ease review and debugging. And of course some unit-tests would be great!

Thanks for the review. Will definitely go over the comments and update accordingly. Just FYI, I have been using http_listener_v2 input plugin as a reference to write the code. The unit tests will come soon as well

srebhan · 2024-10-09T07:49:14Z

Yeah we do have some older code we didn't adapt yet but things changed both on the golang side as well as on us being more strict with the way things are done. ;-) No worries, we will figure it out together. :-)

srebhan · 2024-10-10T11:03:03Z

@syedmhashim I'm loosing a bit the track in the discussion above, could you please push an update with the stuff that is clear and then we discuss the unclear parts?

syedmhashim · 2024-10-11T13:25:15Z

@srebhan Hey! Just pushed some changes and commented "done" on the threads that I resolved. Sorry for the delay. I got occupied with my job

syedmhashim and others added 2 commits October 7, 2024 15:34

feat(inputs/firehose): add input plugin for AWS Data Firehose

00db15c

Merge pull request #1 from syedmhashim/firehose-input-plugin

b2afcbe

feat(inputs/firehose): add input plugin for AWS Data Firehose

telegraf-tiger bot added the feat Improvement on an existing feature such as adding a new setting/mode to an existing plugin label Oct 7, 2024

srebhan changed the title ~~feat(inputs/firehose): add input plugin for AWS Data Firehose~~ feat(inputs.firehose): Add new plugin Oct 8, 2024

telegraf-tiger bot added the plugin/input 1. Request for new input plugins 2. Issues/PRs that are related to input plugins label Oct 8, 2024

srebhan reviewed Oct 8, 2024

View reviewed changes

srebhan self-assigned this Oct 8, 2024

feat(inputs/firehose): add input plugin for AWS Data Firehose

ed2e91a

		@@ -0,0 +1,123 @@
		# AWS Data Firehose HTTP Listener Input Plugin

		Firehose HTTP Listener is a service input plugin that listens for metrics sent

	Firehose HTTP Listener is a service input plugin that listens for metrics sent
	This is a service input plugin that listens for metrics sent

feat(inputs.firehose): Add new plugin #15988

Are you sure you want to change the base?

feat(inputs.firehose): Add new plugin #15988

Conversation

syedmhashim commented Oct 7, 2024

Summary

Checklist

Related issues

telegraf-tiger bot commented Oct 7, 2024

Artifact URLs

srebhan left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

syedmhashim Oct 9, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

syedmhashim Oct 11, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

srebhan Oct 10, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

syedmhashim commented Oct 9, 2024

srebhan commented Oct 9, 2024

srebhan commented Oct 10, 2024

syedmhashim commented Oct 11, 2024

syedmhashim Oct 9, 2024 •

edited

Loading

syedmhashim Oct 11, 2024 •

edited

Loading

srebhan Oct 10, 2024 •

edited

Loading