[RORDEV-1263] Data stream support in ROR audit #1076

mateuszkp96 · 2025-02-12T23:29:57Z

No description provided.

mateuszkp96 · 2025-02-18T20:49:27Z

core/src/main/scala/tech/beshu/ror/accesscontrol/audit/DataStreamAuditSinkCreator.scala

+      _ <- Task.delay(logger.info(s"Trying to setup ROR audit sink with data stream ${settings.dataStreamName.show}.."))
+      _ <- service.createIndexLifecyclePolicy(settings.lifecyclePolicy, settings.lifecyclePolicyJson(service.capabilities.ilmMaxPrimaryShardSize))
+      _ <- service.createComponentTemplateForMappings(settings.componentTemplateMappingsId, settings.mappings, settings.mappingsMetadata)
+      _ <- service.createComponentTemplateForIndex(settings.componentTemplateSettingsId, settings.lifecyclePolicy, settings.indexSettingsMetadata)
+      _ <- service.createIndexTemplate(settings.indexTemplate, settings.dataStreamName, NonEmptyList.of(settings.componentTemplateMappingsId, settings.componentTemplateSettingsId), settings.indexTemplateMetadata)
+      _ <- service.createDataStream(settings.dataStreamName)
+      _ <- Task.delay(logger.info(s"ROR audit data stream ${settings.dataStreamName.show} created."))


Here, we should handle ACKs from service responses. We will probably need to query the resources to ensure they are available. Some of the queries would require new methods based on reflection.
Now, if the request fails on any step, it cannot be resumed (we need something like creating resources if not exist for every one of the given resources)

I'm not sure if I understand the problem. Could you please explain deeply?

Let's assume that creating an index lifecycle request returned the following response:

{ "acknowledged": false }

This means that the master node did not confirm the request. As a result, the index lifecycle policy was not successfully registered in the cluster.
In the next request, we attempt to create a component template that references this lifecycle policy. However, since the master node is unaware of the index lifecycle policy, the request fails.

If you retry to save ROR settings again, you send a creation request for the lifecycle policy and you can get a conflict. The policy name was created in a previous attempt. This leaves us stuck with an index lifecycle policy but without a data stream.

To avoid this, we should proceed with further requests only after the lifecycle policy is confirmed by the master node (we should check the acknowledge response). Or if we resume the creation process, we should skip the request if the lifecycle policy already exists.

yes, IMO it's very important that all of these resource creation methods are idempotent. If the resource (eg. ilm policy exists, we should not create one).

core/src/main/scala/tech/beshu/ror/accesscontrol/audit/AuditingTool.scala

core/src/main/scala/tech/beshu/ror/accesscontrol/domain/indices.scala

core/src/main/scala/tech/beshu/ror/es/AuditSinkService.scala

core/src/main/scala/tech/beshu/ror/es/DataStreamService.scala

core/src/main/scala/tech/beshu/ror/accesscontrol/audit/DataStreamAuditSinkCreator.scala

coutoPL · 2025-02-23T17:57:31Z

core/src/main/scala/tech/beshu/ror/accesscontrol/factory/decoders/AuditingSettingsDecoder.scala

+          .leftMap {
+            case RorAuditDataStream.CreationError.FormatError(msg) =>
+              AuditingSettingsCreationError(Message(
+                s"Illegal format for 'data_stream' - ${msg.show}"


let's add an info that is the ROR audit data stream

coutoPL · 2025-02-23T17:59:27Z

core/src/main/scala/tech/beshu/ror/accesscontrol/factory/decoders/AuditingSettingsDecoder.scala

+      case other =>
+        unsupportedOutputTypeError(
+          unsupportedType = other,
+          supportedTypes = NonEmptyList.of("data_stream", "index", "log")


should we take info consideration the es version also here (I mean when we create the supportedTypes list)

coutoPL · 2025-02-23T18:02:39Z

core/src/main/scala/tech/beshu/ror/boot/ReadonlyRest.scala

@@ -263,21 +266,28 @@ class ReadonlyRest(coreFactory: CoreFactory,
 }

 object ReadonlyRest {
-  type AuditSinkCreator = AuditCluster => AuditSinkService
+  trait AuditSinkCreator {


I wonder if this trait should be defined in the tech.beshu.ror.accesscontrol.audit package. WDYT?

coutoPL · 2025-02-23T18:04:10Z

core/src/main/scala/tech/beshu/ror/boot/ReadonlyRest.scala

    core.rorConfig.auditingSettings
-      .flatMap(settings => AuditingTool.create(settings, auditSinkCreator)(environmentConfig.clock, loggingContext))
+      .map(settings => AuditingTool.create(settings, auditSinkCreator.index, auditSinkCreator.dataStream)(using environmentConfig.clock, loggingContext))


maybe we could pass here AuditSinkCreator instead of extracting data stream and index creators at this level. WDYT?

coutoPL · 2025-02-23T18:05:31Z

core/src/main/scala/tech/beshu/ror/utils/DurationOps.scala

@@ -28,13 +28,28 @@ object DurationOps {
  implicit class RefinedDurationOps(val duration: Duration) extends AnyVal {
    def toRefinedPositive: Either[String, PositiveFiniteDuration] = duration match {
      case v: FiniteDuration if v.toMillis > 0 =>
+        v.toCoarsest


coutoPL · 2025-02-23T18:09:14Z

core/src/test/scala/tech/beshu/ror/unit/acl/logging/AuditingToolTests.scala

@@ -61,39 +61,43 @@ class AuditingToolTests extends AnyWordSpec with MockFactory with BeforeAndAfter
          "request was allowed and verbosity level was ERROR" in {
            val auditingTool = AuditingTool.create(
              settings = auditSettings(new DefaultAuditLogSerializer),
-              auditSinkServiceCreator = _ => mock[AuditSinkService]
-            ).get
+              indexAuditSinkCreator = _ => mock[IndexBasedAuditSinkService],


shouldn't we add datastream-related tests in this suite?

coutoPL · 2025-02-23T18:09:47Z

core/src/test/scala/tech/beshu/ror/unit/boot/ReadonlyRestStartingTests.scala

@@ -1354,16 +1352,16 @@ class ReadonlyRestStartingTests
        mapWithIntervalFrom(refreshInterval) ++
          mapWithMaxYamlSize(maxYamlSize) ++
          Map(
-            "com.readonlyrest.settings.loading.delay" -> "0"
+            "com.readonlyrest.settings.loading.delay" -> "1"


is it intentional?

It's intentional. The FiniteDuration type has been changed to the PositiveFiniteDuration. The value of 0 became invalid, so the default was used (5 seconds) in the tests. With this change, I reduced all unit test execution time by 5 minutes.

coutoPL · 2025-02-23T18:10:13Z

core/src/test/scala/tech/beshu/ror/unit/boot/RorIndexTest.scala

-            "com.readonlyrest.settings.loading.delay" -> "0"
-          )
+        Map(
+          "com.readonlyrest.settings.loading.delay" -> "1"


the same here

The same here

coutoPL · 2025-02-23T18:12:21Z

es67x/src/main/scala/tech/beshu/ror/es/IndexLevelActionFilter.scala

-      new EsAuditSinkService(client)
-    case remote: AuditCluster.RemoteAuditCluster =>
-      HighLevelClientAuditSinkService.create(remote)
+  private def auditSinkCreator: AuditSinkCreator = new AuditSinkCreator {


Maybe instead of throwing the illegal state ex, we can make this state impossible?
Eg. by having two AuditSinkCreator traits? One supporting only index sink. And the second one supporting both?

coutoPL · 2025-02-23T18:22:18Z

es717x/src/main/scala/tech/beshu/ror/es/services/EsAuditSinkService.scala

-class EsAuditSinkService(client: Client,
-                         threadPool: ThreadPool)
-  extends AuditSinkService
+final class EsAuditSinkService(client: NodeClient, threadPool: ThreadPool, jsonParserFactory: XContentJsonParserFactory)


I wonder if we shouldn't rename it to NodeClientBasedAuditSinkService. WDYT?

mateuszkp96 added 4 commits February 13, 2025 00:25

Data stream support in ROR audit

f7d67bd

Tests fixes

46a2b2d

Tests fixes

78296a5

Data stream name format validation

9804323

mateuszkp96 changed the title ~~DRAFT: [RORDEV-1263] Data stream support in ROR audit~~ [RORDEV-1263] Data stream support in ROR audit Feb 18, 2025

mateuszkp96 commented Feb 18, 2025

View reviewed changes

mateuszkp96 requested a review from coutoPL February 18, 2025 21:00

coutoPL requested changes Feb 19, 2025

View reviewed changes

mateuszkp96 added 12 commits February 20, 2025 20:34

Review fixes

47939ec

Fixes

8d05d99

Lifecycle policy json at module level

676ef35

Dynamic mappings

9427943

Dynamic mappings cleanup & refactoring

38bf368

Cleanup

ea03994

EsEnvProvider

44e764a

Protected defs in subclasses

62b7dbf

Refactoring

cc94a1e

Tests fixes

2d4f5d7

Merge branch 'develop' into RORDEV-1263

bc1f22b

Test fix

b66bea7

mateuszkp96 requested a review from coutoPL February 23, 2025 14:34

coutoPL requested changes Feb 23, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[RORDEV-1263] Data stream support in ROR audit #1076

[RORDEV-1263] Data stream support in ROR audit #1076

mateuszkp96 commented Feb 12, 2025

mateuszkp96 Feb 18, 2025

coutoPL Feb 19, 2025

mateuszkp96 Feb 23, 2025

coutoPL Feb 23, 2025

coutoPL Feb 23, 2025

coutoPL Feb 23, 2025

coutoPL Feb 23, 2025

coutoPL Feb 23, 2025

coutoPL Feb 23, 2025

coutoPL Feb 23, 2025

coutoPL Feb 23, 2025

mateuszkp96 Feb 23, 2025

coutoPL Feb 23, 2025

mateuszkp96 Feb 23, 2025

coutoPL Feb 23, 2025

coutoPL Feb 23, 2025

[RORDEV-1263] Data stream support in ROR audit #1076

Are you sure you want to change the base?

[RORDEV-1263] Data stream support in ROR audit #1076

Conversation

mateuszkp96 commented Feb 12, 2025

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