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

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!

plugins/inputs/firehose/README.md

plugins/inputs/firehose/firehose.go

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

srebhan

That looks much better @syedmhashim! Some more cleanups and comments from my side...

plugins/inputs/firehose/README.md

plugins/inputs/firehose/firehose.go

plugins/inputs/firehose/firehose_request.go

syedmhashim · 2024-10-21T09:11:55Z

@srebhan Hi! Pushed some changes and added comments where relevant

srebhan

@syedmhashim thanks for the update. I tried to put more concrete suggestions into my review. Overall, I think we are almost there, the only thing missing are unit-tests with a mocked sender...

srebhan · 2024-10-29T09:48:27Z

plugins/inputs/firehose/firehose.go

+	ParameterTags  []string        `toml:"parameter_tags"`
+
+	tlsint.ServerConfig
+	tlsConf *tls.Config
+
+	once sync.Once
+	Log  telegraf.Logger


Suggested change

ParameterTags []string `toml:"parameter_tags"`

tlsint.ServerConfig

tlsConf *tls.Config

once sync.Once

Log telegraf.Logger

ParameterTags []string `toml:"parameter_tags"`

Log telegraf.Logger `toml:"-"`

tlsint.ServerConfig

tlsConf *tls.Config

once sync.Once

srebhan · 2024-10-29T09:48:47Z

plugins/inputs/firehose/firehose.go

+func (f *Firehose) Gather(_ telegraf.Accumulator) error {
+	return nil
+}


Suggested change

func (f *Firehose) Gather(_ telegraf.Accumulator) error {

return nil

}

func (*Firehose) Gather(telegraf.Accumulator) error {

return nil

}

srebhan · 2024-10-29T09:51:15Z

plugins/inputs/firehose/firehose.go

+		return &Firehose{
+			ServiceAddress: ":8080",
+			Paths:          []string{"/telegraf"},
+		}


Move the timeout initialization here, so the user can overwrite it in the config with whatever he wants

Suggested change

return &Firehose{

ServiceAddress: ":8080",

Paths: []string{"/telegraf"},

}

return &Firehose{

ServiceAddress: ":8080",

Paths: []string{"/telegraf"},

ReadTimeout: config.Duration(time.Second * 5),

WriteTimeout: config.Duration(time.Second * 5),

}

srebhan · 2024-10-29T09:52:32Z

plugins/inputs/firehose/firehose.go

+	if f.ReadTimeout < config.Duration(time.Second) {
+		f.ReadTimeout = config.Duration(time.Second * 5)
+	}
+	if f.WriteTimeout < config.Duration(time.Second) {
+		f.WriteTimeout = config.Duration(time.Second * 5)
+	}


Move timeout to init() (see my suggestion there)... This way, the timeouts are initialized with the defaults but the user can override the values with whatever he/she wants...

Suggested change

if f.ReadTimeout < config.Duration(time.Second) {

f.ReadTimeout = config.Duration(time.Second * 5)

}

if f.WriteTimeout < config.Duration(time.Second) {

f.WriteTimeout = config.Duration(time.Second * 5)

}

srebhan · 2024-10-29T09:54:08Z

plugins/inputs/firehose/firehose.go

+func (f *Firehose) Stop() {
+	err := f.server.Shutdown(context.Background())
+	if err != nil {
+		f.Log.Infof("shutting down server failed: %v", err)


Suggested change

f.Log.Infof("shutting down server failed: %v", err)

f.Log.Errorf("shutting down server failed: %v", err)

plugins/inputs/firehose/firehose_request.go

srebhan · 2024-10-29T10:04:50Z

plugins/inputs/firehose/sample.conf

+  ## Optional access key to accept for authentication.
+  ## AWS Data Firehose uses "x-amz-firehose-access-key" header to set the access key


Suggested change

## Optional access key to accept for authentication.

## AWS Data Firehose uses "x-amz-firehose-access-key" header to set the access key

## Optional access key to accept for authentication.

## AWS Data Firehose uses "x-amz-firehose-access-key" header to set the access key.

## If no access_key is provided (default), authentication is completely disabled and

## this plugin will accept all request ignoring the provided access-key in the request!

plugins/inputs/firehose/firehose_request.go

srebhan · 2024-10-29T10:11:25Z

plugins/inputs/firehose/firehose.go

+	}
+
+	var metrics []telegraf.Metric
+	for _, bytes := range data {


Please don't use bytes as variable name as this is also the name of a built-in Go library. How about sample , msg or record?

srebhan · 2024-10-29T10:15:42Z

plugins/inputs/firehose/firehose.go

+	if err := f.addParameterTags(r, metrics); err != nil {
+		f.Log.Errorf("add parameter tags for request %q failed: %v", requestID, err)
+		r.res.statusCode = http.StatusBadRequest
+		if err = r.sendResponse(res); err != nil {
+			f.Log.Errorf("sending response to request %q failed: %v", requestID, err)
+		}
+		return
+	}
+
+	for _, m := range metrics {
+		m.AddTag("firehose_http_path", req.URL.Path)
+	}
+
+	for _, m := range metrics {
+		f.acc.AddMetric(m)
+	}


How about

Suggested change

if err := f.addParameterTags(r, metrics); err != nil {

f.Log.Errorf("add parameter tags for request %q failed: %v", requestID, err)

r.res.statusCode = http.StatusBadRequest

if err = r.sendResponse(res); err != nil {

f.Log.Errorf("sending response to request %q failed: %v", requestID, err)

}

return

}

for _, m := range metrics {

m.AddTag("firehose_http_path", req.URL.Path)

}

for _, m := range metrics {

f.acc.AddMetric(m)

}

tags, err := r.extractParameterTags(f.ParameterTags)

if err != nil {

f.Log.Errorf("extracting parameter tags for request %q failed: %v", requestID, err)

r.res.statusCode = http.StatusBadRequest

if err = r.sendResponse(res); err != nil {

f.Log.Errorf("sending response to request %q failed: %v", requestID, err)

}

return

}

for _, m := range metrics {

for k, v := range tags {

m.AddTag(k, v)

}

m.AddTag("firehose_http_path", req.URL.Path)

f.acc.AddMetric(m)

}

This way you don't need to iterate over the metrics three times and do have a clear code structure where everything related to processing the request is in the request "class"...

makes sense

telegraf-tiger · 2024-11-05T12:05:59Z

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

Really nice @syedmhashim! Some very small things with the biggest being that log messages should start with a captial letter... The only thing missing are the unit-tests...

srebhan · 2024-11-05T16:31:47Z

plugins/inputs/firehose/firehose.go

+	var err error
+	f.tlsConf, err = f.ServerConfig.TLSConfig()
+	if err != nil {
+		return err
+	}
+
+	return nil


Suggested change

var err error

f.tlsConf, err = f.ServerConfig.TLSConfig()

if err != nil {

return err

}

return nil

var err error

f.tlsConf, err = f.ServerConfig.TLSConfig()

return err

srebhan · 2024-11-05T16:32:34Z

plugins/inputs/firehose/firehose.go

+		}
+	}()
+
+	f.Log.Infof("listening on %s", f.listener.Addr().String())


Suggested change

f.Log.Infof("listening on %s", f.listener.Addr().String())

f.Log.Infof("Listening on %s", f.listener.Addr().String())

srebhan · 2024-11-05T16:32:56Z

plugins/inputs/firehose/firehose.go

+	err := f.server.Shutdown(context.Background())
+	if err != nil {
+		f.Log.Errorf("shutting down server failed: %v", err)
+	}


Suggested change

err := f.server.Shutdown(context.Background())

if err != nil {

f.Log.Errorf("shutting down server failed: %v", err)

}

if err := f.server.Shutdown(context.Background()); err != nil {

f.Log.Errorf("Shutting down server failed: %v", err)

}

srebhan · 2024-11-05T16:34:50Z

plugins/inputs/firehose/firehose.go

+		r.res.statusCode = http.StatusBadRequest
+		f.Log.Errorf("x-amz-firehose-request-id header is not set")
+		if err := r.sendResponse(res); err != nil {
+			f.Log.Errorf("sending response failed: %v", err)


For log-messages please capitalize the message for better readability for the user! Same for all other f.Log calls...

srebhan · 2024-11-05T16:37:19Z

plugins/inputs/firehose/firehose.go

+		return
+	}
+
+	r := &request{req: req}


Would it make sense to set the response status-code to a sensible default here? E.g. 500 internal server error so we catch the cases where we forgot to set a status-code? This would make things future proof. :-)

srebhan · 2024-11-05T16:38:25Z

plugins/inputs/firehose/firehose.go

+	paramTags, err := r.extractParameterTags(f.ParameterTags)
+	if err != nil {
+		f.Log.Errorf("extracting parameter tags for request %q failed: %v", requestID, err)
+		r.res.statusCode = http.StatusBadRequest


Why not setting the status-code in extractParameterTags like you do in the other functions?

srebhan · 2024-11-05T16:39:33Z

plugins/inputs/firehose/firehose.go

+			ServiceAddress: ":8080",
+			Paths:          []string{"/telegraf"},


Please remove those here as you handle them in Init()

Suggested change

ServiceAddress: ":8080",

Paths: []string{"/telegraf"},

Sure. Just a question. What's the difference between initialising the attributes in init()and in Init() and how do we decide which option to choose?

The init() (lowercase) function will be called when importing the package by Go itself. So what you do here is how the structure is registered and that's how each instance looks like before parsing any config options! The Init() (uppercase) function is called after setting the struct fields from the config options. So you should do everything in Init() if the nil value is invalid for the type and option (like strings etc). However, if you do have for example timeouts where a zero-value might mean "wait forever" it is better to set this in init() (lowercase) as the user might want to overwrite the default with zero. Does that make sense?

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

srebhan reviewed Oct 16, 2024

View reviewed changes

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

7afdecc

syedmhashim added 2 commits October 25, 2024 17:48

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

377929b

feat(inputs/firehose): update

737c37e

srebhan reviewed Oct 29, 2024

View reviewed changes

feat(inputs/firehose): update

d6914ba

syedmhashim force-pushed the master branch from 3241fdc to d6914ba Compare November 5, 2024 11:38

srebhan reviewed Nov 5, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

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

syedmhashim commented Oct 7, 2024

srebhan left a comment

syedmhashim commented Oct 9, 2024

srebhan commented Oct 9, 2024

srebhan commented Oct 10, 2024

syedmhashim commented Oct 11, 2024

srebhan left a comment

syedmhashim commented Oct 21, 2024

srebhan left a comment

srebhan Oct 29, 2024

srebhan Oct 29, 2024

srebhan Oct 29, 2024

srebhan Oct 29, 2024

srebhan Oct 29, 2024

srebhan Oct 29, 2024

srebhan Oct 29, 2024

srebhan Oct 29, 2024

syedmhashim Nov 5, 2024

telegraf-tiger bot commented Nov 5, 2024

Artifact URLs

srebhan left a comment

srebhan Nov 5, 2024

srebhan Nov 5, 2024

srebhan Nov 5, 2024

srebhan Nov 5, 2024

srebhan Nov 5, 2024

srebhan Nov 5, 2024

srebhan Nov 5, 2024

syedmhashim Nov 6, 2024

srebhan Nov 11, 2024

	f.Log.Infof("shutting down server failed: %v", err)
	f.Log.Errorf("shutting down server failed: %v", err)

		## Optional access key to accept for authentication.
		## AWS Data Firehose uses "x-amz-firehose-access-key" header to set the access key

	f.Log.Infof("listening on %s", f.listener.Addr().String())
	f.Log.Infof("Listening on %s", f.listener.Addr().String())

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

srebhan left a 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

srebhan left a comment

Choose a reason for hiding this comment

syedmhashim commented Oct 21, 2024

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

telegraf-tiger bot commented Nov 5, 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