[filebeat] Elasticsearch state storage for httpjson and cel inputs #41446

aleksmaus · 2024-10-24T20:38:53Z

Proposed commit message

[filebeat] Elasticsearch state storage for httpjson input

This is a POC for Elasticsearch as State Store Backend for Security Integrations for Agentless solution.

The scope of this change was narrowed down to supporting only httpjson inputs in order to support Okta integration for the initial release. All the other integrations inputs still use the file storage as before.
This is a short term solution for the state storage for k8s environment.

This is the first cut and the details can change depending on the feedback.

Current feature currently could be enabled AGENTLESS_ELASTICSEARCH_STATE_STORE_ENABLED, to be decided how this would be configurable in k8s.

This change currently contains the hacky approach to the AGENTLESS_ELASTICSEARCH_APIKEY overwrite. This allows to the user to provide the ApiKey with elevated permissions that are required in order to be able to create/write/read the state index per input. THIS IS FOR DEVELOPMENT/TESTING ONLY. REMOVE BEFORE THE MERGE.

The existing code relied on the inputs state storage to be fully configurable before the main beat managers runs. The change delays the configuration of httpjson input to the time when the actual configuration is received from the Agent.

There is an assumption that the index template for the state storage indices is already in place before the storage is used

PUT _index_template/agentless_state_template
{
  "index_patterns": [
    "agentless-state-*"
  ],
  "priority": 300,
  "template": {
    "mappings": {
      "properties": {
        "v": {
          "type": "object",
          "enabled": false
        },
        "updated_at": {
          "type": "date",
          "format": "strict_date_optional_time||epoch_millis"
        }
      }
    },
    "settings": {
      "number_of_shards": 1
    }
  }
}

Example of the state storage index content for Okta integration:

{
  "took": 6,
  "timed_out": false,
  "_shards": {
    "total": 1,
    "successful": 1,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": {
      "value": 1,
      "relation": "eq"
    },
    "max_score": 1,
    "hits": [
      {
        "_index": "agentless-state-httpjson-okta.system-028ecf4b-babe-44c6-939e-9e3096af6959",
        "_id": "httpjson::httpjson-okta.system-028ecf4b-babe-44c6-939e-9e3096af6959::https://dev-36006609.okta.com/api/v1/logs",
        "_seq_no": 39,
        "_primary_term": 1,
        "_score": 1,
        "_source": {
          "v": {
            "ttl": 1800000000000,
            "updated": "2024-10-24T20:21:22.032Z",
            "cursor": {
              "published": "2024-10-24T20:19:53.542Z"
            }
          }
        }
      }
    ]
  }
}

The naming convention for all state store is agentless-state-<input id>, since the expectation for agentless we would have only one agent per policy and the agents are ephemeral.

Currently in order to run the agent with Elasticsearch state storage a couple of environment variables would be required:

sudo AGENTLESS_ELASTICSEARCH_STATE_STORE_ENABLED=1 AGENTLESS_ELASTICSEARCH_APIKEY=xxxxxxxx-xvpDXfB:jVMRsW7SRIxxxxxxxxx ./elastic-agent -e

where the ApiKey in the

DEPENDENCIES / TODOS:

Approval of teams for this approach
Kibana (?) side change is required for the agentless-state index template boostrapping
Kibana or the intergration package (or both) change is required in order to include the permissions for agentless-state- with the Elasticsearch ApiKey (Remove the hack). I suspect that Kibana fleet code could be modified to recognize agentless supporting integration and include the proper index name for the agentless-state for the ApiKey permissions.

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.

Disruptive User Impact

The change should have no impact, and without the feature enabled the filebeat should work as before using the file system storage for the state.

mergify · 2024-10-24T20:39:42Z

This pull request does not have a backport label.
If this is a bug or security fix, could you label this PR @aleksmaus? 🙏.
For such, you'll need to label your PR with:

The upcoming major version of the Elastic Stack
The upcoming minor version of the Elastic Stack (if you're not pushing a breaking change)

To fixup this pull request, you need to add the backport labels for the needed
branches, such as:

backport-8./d is the label to automatically backport to the 8./d branch. /d is the digit

mergify · 2024-10-24T20:39:43Z

backport-8.x has been added to help with the transition to the new branch 8.x.
If you don't need it please use backport-skip label and remove the backport-8.x label.

filebeat/beater/filebeat.go

libbeat/statestore/backend/es/store.go

aleksmaus · 2024-10-31T11:49:50Z

@belimawr @cmacknz (or whoever wants/have time to be involved)
I need your feedback on this draft, if this approach is something that we could eventually merge (the ApiKey workaround will be removed once we adjust the kibana fleet).
I think this is an ok solution given the circumstances:

This is fully backwards compatible. If the feature is not enabled, everything would work as before.
The only inputs that are enabled for Elasticsearch backed state storage are the httpjson and cel. We only enabling the limited number of integration relying on httpjson or cel inputs for the first release.
The state initialization for the inputs is delayed until we get the configuration only if the feature is enabled for the input.
The agent logs monitoring will still use the local storage, since we agreed that loosing the agent log when the pod relocated is acceptable.

cmacknz · 2024-11-01T20:23:50Z

@leehinman I'd appreciate a review here to make sure this can co-exist with Beats receivers in agent since that would be the long term way we plan to run agentless inputs.

filebeat/beater/filebeat.go

cmacknz · 2024-11-01T20:29:52Z

filebeat/features/features.go

+
+// List of input types Elasticsearch state store is enabled for
+var esTypesEnabled = map[string]void{
+	"httpjson": {},


Can this be configuration instead of in the code, maybe another env var?

Sure can do. Something like this?
AGENTLESS_ELASTICSEARCH_STATE_STORE_INPUT_TYPES=httpjson,cel

libbeat/statestore/backend/es/store.go

elasticmachine · 2024-11-05T18:22:00Z

Pinging @elastic/elastic-agent-data-plane (Team:Elastic-Agent-Data-Plane)

orestisfl · 2024-11-07T14:23:54Z

filebeat/beater/store.go

+		err error
+
+		esreg    *es.Registry
+		notifier *es.Notifier


(Optional) pointers to potentially nil objects are common places where bugs could be introduced. How about an interface with two implementations, one no-op and one the actual notifier.

filebeat/registrar/registrar.go

libbeat/statestore/backend/es/store.go

filebeat/beater/store.go

aleksmaus

Addressed code review comments

libbeat/statestore/backend/es/store.go

filebeat/beater/store.go

filebeat/registrar/registrar.go

orestisfl · 2024-11-13T11:53:06Z

libbeat/statestore/backend/es/registry.go

+func New(ctx context.Context, log *logp.Logger, notifier *Notifier) (*Registry, error) {
+	r := &Registry{
+		ctx:      ctx,
+		log:      log,
+		notifier: notifier,
+	}
+
+	return r, nil
+}


error return can be removed

Will leave for now as is, as it's consistent with memlog.New that returns the registry and in case if further changes that could return error from the constructor

This reverts commit 34003d4.

…asticsearch state store is enabled if any of the inputs are with AGENTLESS_ELASTICSEARCH_STATE_STORE_INPUT_TYPES.

aleksmaus · 2024-11-25T16:42:19Z

Eliminated AGENTLESS_ELASTICSEARCH_STATE_STORE_ENABLED environment variable setting, it seemed a bit redundant.
Now Elasticsearch state storage is enabled if AGENTLESS_ELASTICSEARCH_STATE_STORE_INPUT_TYPES is set.
Her is an example of shorter command to run the agent now:

sudo AGENTLESS_ELASTICSEARCH_STATE_STORE_INPUT_TYPES="httpjson,cel" AGENTLESS_ELASTICSEARCH_APIKEY=fsOitZIBVlcA-mvxxxxx:jVMRsW7SRIOc-U6VHxxxxx ./elastic-agent -e

It will be event shorter when the APIKEY permissions are properly set through the policy and we can remove that workaround with APIKEY injection.

aleksmaus · 2024-12-02T16:08:27Z

Can I get some reviews on this PR? What is acceptable? What is missing?

Assuming that this "hack" https://github.com/elastic/beats/pull/41446/files#diff-97b65fad069be1072219d7ae6a1e8f64d287e8cb6c1f4e424c801a22200db104R318
will be removed in the final version, when we have the proper permissions changes on Kibana side.

cmacknz

The biggest thing missing now is tests. There needs to be an integration test proving this actually works, ideally against ES running in a container.

I would also like to see some tests that drive the Beat configuration as if it were managed by Elastic agent. There are a couple that start Filebeat with management enabled but using a gRPC server that is under the test control instead of the real agent, they also run with ES running already. https://github.com/cmacknz/beats/blob/d4fc5cdeff6d2bc382f0341fe9118157f6fcb849/x-pack/filebeat/tests/integration/managerV2_test.go#L56-L70

cmacknz · 2024-12-02T21:28:21Z

libbeat/statestore/backend/es/notifier.go

+
+type OnConfigUpdateFunc func(c *conf.C)
+
+type Notifier struct {


This needs tests

cmacknz · 2024-12-02T21:29:16Z

libbeat/statestore/backend/es/notifier.go

+func (n *Notifier) Unsubscribe(id int) {
+	n.mx.Lock()
+	defer n.mx.Unlock()
+	delete(n.listeners, id)


Why does this decrement n.counter?

counter here is just the id (handle) that is returned to the user on Subscribe in order to allow to that "handle" later for Unsubscribe.

cmacknz · 2024-12-02T21:37:38Z

libbeat/statestore/backend/es/store.go

+	}
+	s.cliErr = nil
+
+	cli, err := eslegclient.NewConnectedClient(ctx, c, s.name)


Does this mean there is a client and connection per input?

Per instance of the storage, yes. Do we want to make it singleton/global?

mainly because of this piece:

beats/libbeat/esleg/eslegclient/connection.go

Line 253 in 09e540c

// NewConnectedClient returns a non-thread-safe connection. Make sure for each goroutine you initialize a new connection.

// NewConnectedClient returns a non-thread-safe connection. Make sure for each goroutine you initialize a new connection. func NewConnectedClient(ctx context.Context, cfg *cfg.C, beatname string) (*Connection, error) {

cmacknz · 2024-12-02T21:38:49Z

libbeat/statestore/backend/backend.go

+
+	// Sets the store ID when the full input configuration is acquired
+	// This is needed in order to support Elasticsearch state store naming convention based on the input ID
+	SetID(id string)


The comment implies this means there will be a store per input, would this make more sense being a required argument when the store is constructed? You can't have an interface that includes a constructor, but that is where it feels like it belongs.

Trying to workaround the current code interfaces that are already in place here. That's why had to add additional function to be able to set the ID when the input configuration is actually received and the ID is available.

libbeat/statestore/backend/es/store.go

cmacknz · 2024-12-02T21:49:40Z

libbeat/statestore/backend/es/store.go

+		"query": map[string]any{
+			"match_all": map[string]any{},
+		},
+		"size": 1000, // TODO: we might have to do scroll if there are more than 1000 keys


How likely are we to have more than 1000 keys? The inputs wouldn't have had any kind of restriction on the number of keys before, so hard to tell if this won't matter for a while or is critically important to fix.

From the input types that I'm testing with so far is very very unlikely. For example for httpjson input, it's usually one cursor key/value per stream.

mergify · 2024-12-03T12:52:54Z

This pull request is now in conflicts. Could you fix it? 🙏
To fixup this pull request, you can check out it locally. See documentation: https://help.github.com/articles/checking-out-pull-requests-locally/

git fetch upstream
git checkout -b poc/es_state_store upstream/poc/es_state_store
git merge upstream/main
git push upstream poc/es_state_store

filebeat/beater/filebeat.go

mergify · 2024-12-10T22:30:15Z

This pull request is now in conflicts. Could you fix it? 🙏
To fixup this pull request, you can check out it locally. See documentation: https://help.github.com/articles/checking-out-pull-requests-locally/

git fetch upstream
git checkout -b poc/es_state_store upstream/poc/es_state_store
git merge upstream/main
git push upstream poc/es_state_store

[filebeat] Elasticsearch state storage for httpjson input

55c72d3

aleksmaus added the enhancement label Oct 24, 2024

aleksmaus requested review from orestisfl, olegsu and oren-zohar October 24, 2024 20:38

aleksmaus self-assigned this Oct 24, 2024

botelastic bot added the needs_team Indicates that the issue/PR needs a Team:* label label Oct 24, 2024

aleksmaus added the Team:Security-Deployment and Devices Deployment and Devices Team in Security Solution label Oct 24, 2024

botelastic bot removed the needs_team Indicates that the issue/PR needs a Team:* label label Oct 24, 2024

mergify bot added the backport-8.x Automated backport to the 8.x branch with mergify label Oct 24, 2024

aleksmaus commented Oct 24, 2024

View reviewed changes

filebeat/beater/filebeat.go Show resolved Hide resolved

aleksmaus added 3 commits October 24, 2024 16:54

Fixup tests

1bf288d

Linter

e003053

Enabled elastisearch storage support for cel input and some cleanup

dfce978

aleksmaus commented Oct 29, 2024

View reviewed changes

libbeat/statestore/backend/es/store.go Outdated Show resolved Hide resolved

Remove the "hack" with .Each implementation

e2e25fa

aleksmaus changed the title ~~[filebeat] Elasticsearch state storage for httpjson input~~ [filebeat] Elasticsearch state storage for httpjson and cel inputs Oct 30, 2024

aleksmaus added 2 commits October 31, 2024 07:27

Merge branch 'main' into poc/es_state_store

953355b

Adjust for the latest main es client signature change

c1fc2a8

aleksmaus added 3 commits October 31, 2024 08:02

Make check happy

c9b0256

Fixed missing interface method on test mock store

21d451d

Add error check in ES store Each

ffb9364

cmacknz requested a review from leehinman November 1, 2024 20:23

cmacknz reviewed Nov 1, 2024

View reviewed changes

filebeat/beater/filebeat.go Show resolved Hide resolved

cmacknz reviewed Nov 1, 2024

View reviewed changes

libbeat/statestore/backend/es/store.go Show resolved Hide resolved

pierrehilbert added the Team:Elastic-Agent-Data-Plane Label for the Agent Data Plane team label Nov 5, 2024

Delete the dev tests file

24000d7

orestisfl reviewed Nov 7, 2024

View reviewed changes

Address code review comments

a0d019e

aleksmaus commented Nov 7, 2024

View reviewed changes

Add refresh=wait_for for data consistency

34003d4

orestisfl reviewed Nov 13, 2024

View reviewed changes

aleksmaus added 4 commits November 18, 2024 08:28

Revert "Add refresh=wait_for for data consistency"

36e9983

This reverts commit 34003d4.

Cleanup

aee4112

Added updated_at field

76b11a4

Merge branch 'main' into poc/es_state_store

1fe1c0e

qcorporation mentioned this pull request Nov 21, 2024

[Epic] Road to Agentless + Security Integrations Release (Phase I) elastic/integrations#11810

Open

aleksmaus added 2 commits November 25, 2024 10:16

Merge branch 'main' into poc/es_state_store

3c5bab1

Eliminate AGENTLESS_ELASTICSEARCH_STATE_STORE_ENABLED setting. The El…

4ac1bd0

…asticsearch state store is enabled if any of the inputs are with AGENTLESS_ELASTICSEARCH_STATE_STORE_INPUT_TYPES.

Merge branch 'main' into poc/es_state_store

535560e

cmacknz reviewed Dec 2, 2024

View reviewed changes

Merge branch 'main' into poc/es_state_store

4940097

orestisfl reviewed Dec 9, 2024

View reviewed changes

filebeat/beater/filebeat.go Outdated Show resolved Hide resolved

aleksmaus added 3 commits December 9, 2024 10:21

Added logging per code review request

e73a1b7

Some cleanup and notifier utz

6c39fd0

Add some integration testing coverage

716758f

aleksmaus requested a review from a team as a code owner December 10, 2024 20:59

Merge branch 'main' into poc/es_state_store

5ec7ed4

aleksmaus requested review from cmacknz and orestisfl December 12, 2024 15:18

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[filebeat] Elasticsearch state storage for httpjson and cel inputs #41446

[filebeat] Elasticsearch state storage for httpjson and cel inputs #41446

aleksmaus commented Oct 24, 2024 •

edited

Loading

mergify bot commented Oct 24, 2024

mergify bot commented Oct 24, 2024

aleksmaus commented Oct 31, 2024

cmacknz commented Nov 1, 2024

cmacknz Nov 1, 2024

aleksmaus Nov 4, 2024

elasticmachine commented Nov 5, 2024

orestisfl Nov 7, 2024

aleksmaus left a comment

orestisfl Nov 13, 2024

aleksmaus Nov 13, 2024

aleksmaus commented Nov 25, 2024

aleksmaus commented Dec 2, 2024

cmacknz left a comment

cmacknz Dec 2, 2024

cmacknz Dec 2, 2024

aleksmaus Dec 3, 2024

cmacknz Dec 2, 2024

aleksmaus Dec 3, 2024

aleksmaus Dec 6, 2024

cmacknz Dec 2, 2024

aleksmaus Dec 3, 2024

cmacknz Dec 2, 2024

aleksmaus Dec 3, 2024

mergify bot commented Dec 3, 2024

mergify bot commented Dec 10, 2024


		type OnConfigUpdateFunc func(c *conf.C)

		type Notifier struct {

[filebeat] Elasticsearch state storage for httpjson and cel inputs #41446

Are you sure you want to change the base?

[filebeat] Elasticsearch state storage for httpjson and cel inputs #41446

Conversation

aleksmaus commented Oct 24, 2024 • edited Loading

Proposed commit message

DEPENDENCIES / TODOS:

Checklist

Disruptive User Impact

mergify bot commented Oct 24, 2024

mergify bot commented Oct 24, 2024

aleksmaus commented Oct 31, 2024

cmacknz commented Nov 1, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

elasticmachine commented Nov 5, 2024

Choose a reason for hiding this comment

aleksmaus left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aleksmaus commented Nov 25, 2024

aleksmaus commented Dec 2, 2024

cmacknz 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

mergify bot commented Dec 3, 2024

mergify bot commented Dec 10, 2024

aleksmaus commented Oct 24, 2024 •

edited

Loading