Add cometd input for Salesforce connector #1

kush-elastic · 2021-11-26T07:24:59Z

Enhancement

What does this PR do?

Real-time logs from the Salesforce Streaming API can be collected using the cometD input. For example, live Login & Logout related logs.

Right now with this PR, when the cometd input is enabled, you can start seeing the authentication log messages that are retrieved after successful authentication using the credentials provided as a part of the configuration. You will also be able to see the real-time data being collected from the Salesforce Steaming API.
For configuration:

- type: cometd
  channel_name: /event/LoginEventStream
  auth.oauth2:
    client.id: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
    client.secret: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
    token_url: https://login.salesforce.com/services/oauth2/token
    user: [email protected]
    password: P@$$W0₹D

Why is it important?

It is a prerequisite for the Salesforce Observability Connector for the collection of real-time data from the Streaming APIs.
With all the different logs in cometD from different services, it will be good to have a dedicated Filebeat input to retrieve the same using cometD.

Checklist

My code follows the style guidelines of this project
I have commented my code, particularly in hard-to-understand areas
I have made corresponding changes to the documentation
I have made corresponding change to the default configuration files
I have added tests that prove my fix is effective or that my feature works
I have added an entry in CHANGELOG.next.asciidoc or CHANGELOG-developer.next.asciidoc.

mtojek

General comments

It would great to introduce an integration test. See MQTT input and mosquito server:
https://github.com/elastic/beats/blob/23e4403ae093fcc8f7905345cad2c7ad256976d8/filebeat/input/mqtt/mqtt_integration_test.go
https://github.com/elastic/beats/tree/23e4403ae093fcc8f7905345cad2c7ad256976d8/filebeat/input/mqtt

mtojek · 2021-11-29T11:05:29Z

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

+}
+
+func (in *cometdInput) newPubsubClient(ctx context.Context) (*http.Client, error) {
+


nit: empty line

mtojek · 2021-11-29T11:05:46Z

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

+		httpcommon.WithKeepaliveSettings{Disable: true},
+	)
+	if err != nil {
+		return nil, err


Please use errors.Wrap with context everywhere.

Updated with: error.Errorf with context

mtojek

I suppose you want to keep it here to accumulate also other PRs. What would be the cut-off point, I mean, when do you plan to merge this PR?

mtojek · 2021-12-06T09:03:15Z

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

+		in.workerWg.Add(1)
+		go func() {
+			in.log.Info("Pub/Sub input worker has started.")
+			defer in.log.Info("Pub/Sub input worker has stopped.")


Just double-checking, is the order of these defers important?

mtojek · 2021-12-06T09:05:21Z

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

+
+	// For testing http client
+	// Start of the test code for authentication and client creation
+	baseUrl := "<put base URL here>"


Do you plan to add the baseUrl to the configuration?

mtojek · 2021-12-06T09:06:11Z

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

+		httpcommon.WithKeepaliveSettings{Disable: true},
+	)
+	if err != nil {
+		return nil, fmt.Errorf("cometd client: error on newHTTPClient: %v", err)


nit: a typo in comment? newHTTPClient

mtojek · 2021-12-06T09:08:51Z

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

+	})
+}
+
+// Stop stops the pubsub input and waits for it to fully stop.


AFAIR It should be the other way round: Stop just stops the logic, but doesn't wait, but Wait can call the Stop and wait until all processing is done.

mtojek

First of all, you're mixing the CometD/Bayeux concept and the Salesforce implementation. It doesn't mean that every CometD server requires authorization (any, not just OAuth) or it's Salesforce :) Please take a look at the httpjson implementation to see how it interacts with different auth providers. The other way would be implementing the input as salesforce-cometd, but I see the value in having a generic cometd input, so let's try to make it generic.

Major issues to be adjusted:

The input must support the graceful shutdown.
Input tests are missing. All paths should be covered.
Make the input as stateless as possible. Do not create resources while the input is running if it isn't necessary. It will impact the overall performance.

As this PR include the Filebeat input, please invite also Tiago.

mtojek · 2021-12-24T07:40:53Z

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

+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package cometd


This entire file is missing unit tests. It's a must-have to cover all code paths.

We have covered the base unit test case for the creation of the new input using the NewInput function in input_test.go. We're unsure how we can write the unit test cases for other functions and paths the code falls into, so we'll check that out.

mtojek · 2021-12-24T07:44:04Z

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

+		select {
+		case e := <-in.out:
+			if !e.Successful {
+				if e.Data.Object == nil {


comment please, why e.Data.Object can be nil?

Added the appropriate comment, thanks!

mtojek · 2021-12-24T07:45:12Z

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

+				if err != nil {
+					return fmt.Errorf("JSON error: %v", err)
+				}
+				err = json.Unmarshal(e.Data.Object, &event)


comment please, why json.Marshal, then json.Marshal?

Added the appropriate comment, thanks!

mtojek · 2021-12-24T07:46:51Z

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

+type Status struct {
+	channels  []string
+	clientID  string
+	connected bool
+}
+
+type BayeuxHandshake []struct {
+	ClientID string `json:"clientId"`
+	Channel  string `json:"channel"`
+	Ext      struct {
+		Replay bool `json:"replay"`
+	} `json:"ext"`
+	MinimumVersion           string   `json:"minimumVersion"`
+	Successful               bool     `json:"successful"`
+	SupportedConnectionTypes []string `json:"supportedConnectionTypes"`
+	Version                  string   `json:"version"`
+}
+
+type Subscription struct {
+	ClientID     string `json:"clientId"`
+	Channel      string `json:"channel"`
+	Subscription string `json:"subscription"`
+	Successful   bool   `json:"successful"`
+}
+
+type Credentials struct {
+	AccessToken string `json:"access_token"`
+	InstanceURL string `json:"instance_url"`
+	IssuedAt    string `json:"issued_at"`
+	ID          string `json:"id"`
+	TokenType   string `json:"token_type"`
+	Signature   string `json:"signature"`
+}


Do these values need to be exposed (Status, BayeuxHandshake, Subscription, Credentials)?

Updated it.

mtojek · 2021-12-24T07:47:46Z

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

+	req.Header.Add("Content-Type", "application/json")
+	req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", b.creds.AccessToken))
+	// Passing back cookies is required though undocumented in Salesforce API
+	// We were unable to get process working without passing cookies back to SF server.


I think you can strip this line from here :)

mtojek · 2021-12-24T07:52:05Z

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

+	return sub, nil
+}
+
+func (b *Bayeux) connect(out chan TriggerEvent, log *logp.Logger) (chan TriggerEvent, error) {


This goroutine never finishes. Please adjust the whole input, so it can be gracefully shut down.

We are closing the channel before the input shuts down, so I think it is shutting down gracefully.

mtojek · 2021-12-24T07:52:16Z

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

+		"client_secret": {o.ClientSecret},
+		"username":      {o.User},
+		"password":      {o.Password}}
+	res, err := http.PostForm(route, params)


nit: response

mtojek · 2021-12-24T07:55:29Z

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

+)
+
+var (
+	status = Status{[]string{}, "", false}


What is this? why is it global? What will happen if there are multiple configs enabled?

It is used for storing all the channels that are being subscribed and the client IDs

It has been declared as global in order to store multiple channels

The channel names from all the enabled configs would be stored there

mtojek · 2021-12-24T08:20:05Z

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

+		return nil, fmt.Errorf("bad Call request: %v", err)
+	}
+	req.Header.Add("Content-Type", "application/json")
+	req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", b.creds.AccessToken))


Depending on the implementation, Salesforce oriented or general, this can be optional, right?

We are not quite sure about this one.

mtojek · 2021-12-24T08:28:53Z

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

+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package cometd


I'm not convinced about keeping all the logic in a single file, I think you have to adjust it based on the SoC principle.

Right, we tried following the gcppubsub and that's why we kept it all in a single file.

yug-crest · 2021-12-24T13:37:34Z

First of all, you're mixing the CometD/Bayeux concept and the Salesforce implementation. It doesn't mean that every CometD server requires authorization (any, not just OAuth) or it's Salesforce :) Please take a look at the httpjson implementation to see how it interacts with different auth providers. The other way would be implementing the input as salesforce-cometd, but I see the value in having a generic cometd input, so let's try to make it generic.

Major issues to be adjusted:

The input must support the graceful shutdown.

Input tests are missing. All paths should be covered.

Make the input as stateless as possible. Do not create resources while the input is running if it isn't necessary. It will impact the overall performance.

As this PR include the Filebeat input, please invite also Tiago.

We have tried to make this input generic but at some places like authorization, we had to go Salesforce specific due to the time constraint.
Regarding the graceful shutdown - As far as we tested, it does shut down gracefully.
We are not very sure about points 2 and 3, and how we can incorporate the changes to address them. Can you please elaborate on them?

andresrc

Temporary block

andresrc

Removing the block, review from the team pending.

yug-crest · 2022-01-28T14:40:07Z

Raised a new PR to elastic:master branch. Please refer the same and ignore this one.
Link to the PR: elastic#30089

yug-rajani · 2022-03-24T06:33:45Z

Closing this PR as we have raised a PR in the elastic/beats repo.
Link: elastic#30089

mtojek reviewed Nov 29, 2021

View reviewed changes

kush-elastic marked this pull request as ready for review December 1, 2021 11:40

kush-elastic marked this pull request as draft December 1, 2021 11:41

mtojek reviewed Dec 6, 2021

View reviewed changes

yug-crest changed the title ~~Add authentication mechanism for cometd input for Salesforce connector~~ Add cometd input for Salesforce connector Dec 8, 2021

yug-crest marked this pull request as ready for review December 16, 2021 13:30

yug-crest self-assigned this Dec 24, 2021

mtojek reviewed Dec 24, 2021

View reviewed changes

andresrc suggested changes Jan 10, 2022

View reviewed changes

yug-crest requested a review from mtojek January 19, 2022 11:29

andresrc reviewed Jan 24, 2022

View reviewed changes

yug-rajani closed this Mar 24, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add cometd input for Salesforce connector #1

Add cometd input for Salesforce connector #1

kush-elastic commented Nov 26, 2021 •

edited by yug-crest

Loading

mtojek left a comment

mtojek Nov 29, 2021

kush-elastic Dec 1, 2021

mtojek Nov 29, 2021

kush-elastic Dec 1, 2021

mtojek left a comment

mtojek Dec 6, 2021

mtojek Dec 6, 2021

mtojek Dec 6, 2021

mtojek Dec 6, 2021

mtojek left a comment

mtojek Dec 24, 2021

yug-crest Dec 24, 2021

mtojek Dec 24, 2021

yug-crest Dec 24, 2021

mtojek Dec 24, 2021

yug-crest Dec 24, 2021

mtojek Dec 24, 2021

yug-crest Dec 24, 2021 •

edited

Loading

mtojek Dec 24, 2021

mtojek Dec 24, 2021

yug-crest Dec 24, 2021

mtojek Dec 24, 2021

mtojek Dec 24, 2021

yug-crest Dec 24, 2021

mtojek Dec 24, 2021

yug-crest Dec 24, 2021

mtojek Dec 24, 2021

yug-crest Dec 24, 2021 •

edited

Loading

yug-crest commented Dec 24, 2021

andresrc left a comment

andresrc left a comment

yug-crest commented Jan 28, 2022

yug-rajani commented Mar 24, 2022

		}

		func (in cometdInput) newPubsubClient(ctx context.Context) (http.Client, error) {

Add cometd input for Salesforce connector #1

Add cometd input for Salesforce connector #1

Conversation

kush-elastic commented Nov 26, 2021 • edited by yug-crest Loading

What does this PR do?

Why is it important?

Checklist

mtojek 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

mtojek 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

mtojek 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

yug-crest Dec 24, 2021 • 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

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

yug-crest Dec 24, 2021 • edited Loading

Choose a reason for hiding this comment

yug-crest commented Dec 24, 2021

andresrc left a comment

Choose a reason for hiding this comment

andresrc left a comment

Choose a reason for hiding this comment

yug-crest commented Jan 28, 2022

yug-rajani commented Mar 24, 2022

kush-elastic commented Nov 26, 2021 •

edited by yug-crest

Loading

yug-crest Dec 24, 2021 •

edited

Loading

yug-crest Dec 24, 2021 •

edited

Loading