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

Author: mateuszkp96

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

@@ -18,26 +18,27 @@ package tech.beshu.ror.accesscontrol.audit
 
 import cats.Show
 import cats.data.NonEmptyList
+import cats.implicits.*
 import monix.eval.Task
 import org.apache.logging.log4j.scala.Logging
 import org.json.JSONObject
 import tech.beshu.ror.accesscontrol.audit.AuditingTool.Settings.AuditSink
 import tech.beshu.ror.accesscontrol.audit.AuditingTool.Settings.AuditSink.{Disabled, Enabled}
+import tech.beshu.ror.accesscontrol.audit.sink.*
 import tech.beshu.ror.accesscontrol.blocks.Block.{History, Verbosity}
 import tech.beshu.ror.accesscontrol.blocks.metadata.UserMetadata
 import tech.beshu.ror.accesscontrol.blocks.{Block, BlockContext}
-import tech.beshu.ror.accesscontrol.domain.{AuditCluster, RorAuditIndexTemplate, RorAuditLoggerName}
+import tech.beshu.ror.accesscontrol.domain.{AuditCluster, RorAuditDataStream, RorAuditIndexTemplate, RorAuditLoggerName}
 import tech.beshu.ror.accesscontrol.logging.ResponseContext
 import tech.beshu.ror.accesscontrol.request.RequestContext
 import tech.beshu.ror.audit.instances.DefaultAuditLogSerializer
 import tech.beshu.ror.audit.{AuditLogSerializer, AuditRequestContext, AuditResponseContext}
-import tech.beshu.ror.es.AuditSinkService
 import tech.beshu.ror.implicits.*
 
 import java.time.Clock
 
-class AuditingTool private(auditSinks: NonEmptyList[BaseAuditSink])
-                          (implicit loggingContext: LoggingContext) {
+final class AuditingTool private(auditSinks: NonEmptyList[BaseAuditSink])
+                                (implicit loggingContext: LoggingContext) {
 
   def audit[B <: BlockContext](response: ResponseContext[B]): Task[Unit] = {
     val auditResponseContext = toAuditResponse(response)
@@ -141,18 +142,23 @@ class AuditingTool private(auditSinks: NonEmptyList[BaseAuditSink])
 object AuditingTool extends Logging {
 
   final case class Settings(auditSinks: NonEmptyList[Settings.AuditSink])
+
   object Settings {
 
     sealed trait AuditSink
+
     object AuditSink {
       final case class Enabled(config: AuditSink.Config) extends AuditSink
+
       case object Disabled extends AuditSink
 
       sealed trait Config
+
       object Config {
         final case class EsIndexBasedSink(logSerializer: AuditLogSerializer,
                                           rorAuditIndexTemplate: RorAuditIndexTemplate,
                                           auditCluster: AuditCluster) extends Config
+
         object EsIndexBasedSink {
           val default: EsIndexBasedSink = EsIndexBasedSink(
             logSerializer = new DefaultAuditLogSerializer,
@@ -161,8 +167,21 @@ object AuditingTool extends Logging {
           )
         }
 
+        final case class EsDataStreamBasedSink(logSerializer: AuditLogSerializer,
+                                               rorAuditDataStream: RorAuditDataStream,
+                                               auditCluster: AuditCluster) extends Config
+
+        object EsDataStreamBasedSink {
+          val default: EsDataStreamBasedSink = EsDataStreamBasedSink(
+            logSerializer = new DefaultAuditLogSerializer,
+            rorAuditDataStream = RorAuditDataStream.default,
+            auditCluster = AuditCluster.LocalAuditCluster
+          )
+        }
+
         final case class LogBasedSink(logSerializer: AuditLogSerializer,
                                       loggerName: RorAuditLoggerName) extends Config
+
         object LogBasedSink {
           val default: LogBasedSink = LogBasedSink(
             logSerializer = new DefaultAuditLogSerializer,
@@ -174,16 +193,12 @@ object AuditingTool extends Logging {
   }
 
   def create(settings: Settings,
-             auditSinkServiceCreator: AuditCluster => AuditSinkService)
+             auditSinkServiceCreator: AuditSinkServiceCreator)
             (implicit clock: Clock,
-             loggingContext: LoggingContext): Option[AuditingTool] = {
-    createAuditSinks(settings, auditSinkServiceCreator) match {
+             loggingContext: LoggingContext): Task[Option[AuditingTool]] = {
+    createAuditSinks(settings, auditSinkServiceCreator).map {
       case Some(auditSinks) =>
-        implicit val auditSinkShow: Show[BaseAuditSink] = Show.show {
-          case _: EsIndexBasedAuditSink => "index"
-          case _: LogBasedAuditSink => "log"
-        }
-        logger.info(s"The audit is enabled with the given outputs: [${auditSinks.show}]")
+        logger.info(s"The audit is enabled with the given outputs: [${auditSinks.toList.show}]")
         Some(new AuditingTool(auditSinks))
       case None =>
         logger.info("The audit is disabled because no output is enabled")
@@ -192,20 +207,51 @@ object AuditingTool extends Logging {
   }
 
   private def createAuditSinks(settings: Settings,
-                               auditSinkServiceCreator: AuditCluster => AuditSinkService)
-                              (implicit clock: Clock): Option[NonEmptyList[BaseAuditSink]] = {
+                               auditSinkServiceCreator: AuditSinkServiceCreator)
+                              (using Clock): Task[Option[NonEmptyList[SupportedAuditSink]]] = {
     settings
       .auditSinks
       .toList
-      .flatMap {
-        case Enabled(AuditSink.Config.EsIndexBasedSink(logSerializer, rorAuditIndexTemplate, auditCluster)) =>
-          EsIndexBasedAuditSink(logSerializer, rorAuditIndexTemplate, auditSinkServiceCreator(auditCluster)).some
+      .map[Task[Option[SupportedAuditSink]]] {
+        case Enabled(config: AuditSink.Config.EsIndexBasedSink) =>
+          val serviceCreator: IndexBasedAuditSinkServiceCreator = auditSinkServiceCreator match {
+            case creator: DataStreamAndIndexBasedAuditSinkServiceCreator => creator
+            case creator: IndexBasedAuditSinkServiceCreator => creator
+          }
+          createIndexSink(config, serviceCreator).map(_.some)
+        case Enabled(config: AuditSink.Config.EsDataStreamBasedSink) =>
+          auditSinkServiceCreator match {
+            case creator: DataStreamAndIndexBasedAuditSinkServiceCreator =>
+              createDataStreamSink(config, creator).map(_.some)
+            case _: IndexBasedAuditSinkServiceCreator =>
+              // todo improvement - make this state impossible
+              Task.raiseError(new IllegalStateException("Data stream audit sink is not supported in this version"))
+          }
         case Enabled(AuditSink.Config.LogBasedSink(serializer, loggerName)) =>
-          new LogBasedAuditSink(serializer, loggerName).some
+          Task.delay(new LogBasedAuditSink(serializer, loggerName).some)
         case Disabled =>
-          None
+          Task.pure(None)
       }
-      .toNel
+      .sequence
+      .map(_.flatten.toNel)
   }
 
+  private def createIndexSink(config: AuditSink.Config.EsIndexBasedSink,
+                              serviceCreator: IndexBasedAuditSinkServiceCreator)(using Clock): Task[SupportedAuditSink] = Task.delay {
+    val service = serviceCreator.index(config.auditCluster)
+    EsIndexBasedAuditSink(config.logSerializer, config.rorAuditIndexTemplate, service)
+  }
+
+  private def createDataStreamSink(config: AuditSink.Config.EsDataStreamBasedSink,
+                                   serviceCreator: DataStreamAndIndexBasedAuditSinkServiceCreator): Task[SupportedAuditSink] =
+    Task.delay(serviceCreator.dataStream(config.auditCluster))
+      .flatMap(EsDataStreamBasedAuditSink.create(config.logSerializer, config.rorAuditDataStream, _))
+
+  private type SupportedAuditSink = EsIndexBasedAuditSink | EsDataStreamBasedAuditSink | LogBasedAuditSink
+
+  private given Show[SupportedAuditSink] = Show.show {
+    case _: EsIndexBasedAuditSink => "index"
+    case _: LogBasedAuditSink => "log"
+    case _: EsDataStreamBasedAuditSink => "data_stream"
+  }
 }

core/src/main/scala/tech/beshu/ror/accesscontrol/audit/sink/AuditDataStreamCreator.scala

@@ -0,0 +1,115 @@
+/*
+ *    This file is part of ReadonlyREST.
+ *
+ *    ReadonlyREST is free software: you can redistribute it and/or modify
+ *    it under the terms of the GNU General Public License as published by
+ *    the Free Software Foundation, either version 3 of the License, or
+ *    (at your option) any later version.
+ *
+ *    ReadonlyREST is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU General Public License for more details.
+ *
+ *    You should have received a copy of the GNU General Public License
+ *    along with ReadonlyREST.  If not, see http://www.gnu.org/licenses/
+ */
+package tech.beshu.ror.accesscontrol.audit.sink
+
+import cats.data.NonEmptyList
+import eu.timepit.refined.types.string.NonEmptyString
+import monix.eval.Task
+import org.apache.logging.log4j.scala.Logging
+import tech.beshu.ror.accesscontrol.domain.{DataStreamName, RorAuditDataStream, TemplateName}
+import tech.beshu.ror.es.DataStreamService
+import tech.beshu.ror.es.DataStreamService.DataStreamSettings
+import tech.beshu.ror.es.DataStreamService.DataStreamSettings.*
+import tech.beshu.ror.implicits.*
+import tech.beshu.ror.utils.RefinedUtils.*
+
+import java.util.concurrent.TimeUnit
+
+final class AuditDataStreamCreator(services: NonEmptyList[DataStreamService]) extends Logging {
+
+  def createIfNotExists(dataStreamName: RorAuditDataStream): Task[Unit] = {
+    services.toList.traverse(createIfNotExists(_, dataStreamName)).map((_: List[Unit]) => ())
+  }
+
+  private def createIfNotExists(service: DataStreamService, dataStreamName: RorAuditDataStream): Task[Unit] = {
+    service
+      .checkDataStreamExists(dataStreamName.dataStream)
+      .flatMap {
+        case true =>
+          Task.delay(logger.info(s"Data stream ${dataStreamName.dataStream.show} already exists"))
+        case false =>
+          val settings = defaultSettingsFor(dataStreamName.dataStream)
+          setupDataStream(service, settings)
+      }
+  }
+
+  private def setupDataStream(service: DataStreamService, settings: DataStreamSettings): Task[Unit] = {
+    for {
+      _ <- Task.delay(logger.info(s"Trying to setup ROR audit data stream ${settings.dataStreamName.show} with settings.."))
+      _ <- service.fullySetupDataStream(settings)
+      _ <- Task.delay(logger.info(s"ROR audit data stream ${settings.dataStreamName.show} created."))
+    } yield ()
+  }
+
+  private def defaultSettingsFor(dataStreamName: DataStreamName.Full) = {
+    val defaultLifecyclePolicy = LifecyclePolicy(
+      id = NonEmptyString.unsafeFrom(s"${dataStreamName.value.value}-lifecycle-policy"),
+      hotPhase = LifecyclePolicy.HotPhase(
+        LifecyclePolicy.Rollover(
+          maxAge = positiveFiniteDuration(1, TimeUnit.DAYS),
+          maxPrimaryShardSizeInGb = Some(positiveInt(50))
+        )
+      ),
+      warmPhase = Some(LifecyclePolicy.WarmPhase(
+        minAge = positiveFiniteDuration(14, TimeUnit.DAYS),
+        shrink = Some(LifecyclePolicy.Shrink(numberOfShards = positiveInt(1))),
+        forceMerge = Some(LifecyclePolicy.ForceMerge(maxNumSegments = positiveInt(1)))
+      )),
+      coldPhase = Some(LifecyclePolicy.ColdPhase(
+        minAge = positiveFiniteDuration(30, TimeUnit.DAYS),
+        freeze = true
+      ))
+    )
+
+    val defaultMappings = ComponentTemplateMappings(
+      templateName = templateNameFrom(s"${dataStreamName.value.value}-mappings"),
+      timestampField = "@timestamp",
+      metadata = metadata("Data mappings for ReadonlyREST audit data stream")
+    )
+
+    val settings = ComponentTemplateSettings(
+      templateName = templateNameFrom(s"${dataStreamName.value.value}-settings"),
+      lifecyclePolicyId = defaultLifecyclePolicy.id,
+      metadata = metadata("Index settings for ReadonlyREST audit data stream")
+    )
+
+    val indexTemplate = IndexTemplateSettings(
+      templateName = templateNameFrom(s"${dataStreamName.value.value}-template"),
+      dataStreamName = dataStreamName,
+      componentTemplates = NonEmptyList.of(defaultMappings.templateName, settings.templateName),
+      metadata = metadata("Index template for ReadonlyREST audit data stream")
+    )
+
+    DataStreamSettings(
+      dataStreamName,
+      defaultLifecyclePolicy,
+      defaultMappings,
+      settings,
+      indexTemplate,
+    )
+  }
+
+  private def templateNameFrom(value: String) = {
+    TemplateName
+      .fromString(value)
+      .getOrElse(throw new IllegalStateException("Template name should be non-empty"))
+  }
+
+  private def metadata(description: String) = Map("description" -> description)
+
+}
+

core/src/main/scala/tech/beshu/ror/accesscontrol/audit/sink/AuditSinkServiceCreator.scala

@@ -0,0 +1,37 @@
+/*
+ *    This file is part of ReadonlyREST.
+ *
+ *    ReadonlyREST is free software: you can redistribute it and/or modify
+ *    it under the terms of the GNU General Public License as published by
+ *    the Free Software Foundation, either version 3 of the License, or
+ *    (at your option) any later version.
+ *
+ *    ReadonlyREST is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU General Public License for more details.
+ *
+ *    You should have received a copy of the GNU General Public License
+ *    along with ReadonlyREST.  If not, see http://www.gnu.org/licenses/
+ */
+package tech.beshu.ror.accesscontrol.audit.sink
+
+import tech.beshu.ror.accesscontrol.domain.AuditCluster
+import tech.beshu.ror.es.{DataStreamBasedAuditSinkService, IndexBasedAuditSinkService}
+
+sealed trait AuditSinkServiceCreator
+
+trait IndexBasedAuditSinkServiceCreator
+  extends AuditSinkServiceCreator {
+
+  def index(cluster: AuditCluster): IndexBasedAuditSinkService
+}
+
+trait DataStreamAndIndexBasedAuditSinkServiceCreator
+  extends AuditSinkServiceCreator
+    with IndexBasedAuditSinkServiceCreator {
+
+  def dataStream(cluster: AuditCluster): DataStreamBasedAuditSinkService
+}
+
+

core/src/main/scala/tech/beshu/ror/accesscontrol/audit/BaseAuditSink.scala renamed to core/src/main/scala/tech/beshu/ror/accesscontrol/audit/sink/BaseAuditSink.scala

@@ -14,7 +14,7 @@
  *    You should have received a copy of the GNU General Public License
  *    along with ReadonlyREST.  If not, see http://www.gnu.org/licenses/
  */
-package tech.beshu.ror.accesscontrol.audit
+package tech.beshu.ror.accesscontrol.audit.sink
 
 import monix.eval.Task
 import org.json.JSONObject

core/src/main/scala/tech/beshu/ror/accesscontrol/audit/sink/EsDataStreamBasedAuditSink.scala

@@ -0,0 +1,51 @@
+/*
+ *    This file is part of ReadonlyREST.
+ *
+ *    ReadonlyREST is free software: you can redistribute it and/or modify
+ *    it under the terms of the GNU General Public License as published by
+ *    the Free Software Foundation, either version 3 of the License, or
+ *    (at your option) any later version.
+ *
+ *    ReadonlyREST is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU General Public License for more details.
+ *
+ *    You should have received a copy of the GNU General Public License
+ *    along with ReadonlyREST.  If not, see http://www.gnu.org/licenses/
+ */
+package tech.beshu.ror.accesscontrol.audit.sink
+
+import monix.eval.Task
+import org.json.JSONObject
+import tech.beshu.ror.accesscontrol.domain.RorAuditDataStream
+import tech.beshu.ror.audit.{AuditLogSerializer, AuditResponseContext}
+import tech.beshu.ror.es.DataStreamBasedAuditSinkService
+
+private[audit] final class EsDataStreamBasedAuditSink private(serializer: AuditLogSerializer,
+                                                              rorAuditDataStream: RorAuditDataStream,
+                                                              auditSinkService: DataStreamBasedAuditSinkService)
+  extends BaseAuditSink(serializer) {
+
+  override protected def submit(event: AuditResponseContext, serializedEvent: JSONObject): Task[Unit] = Task {
+    auditSinkService.submit(
+      dataStreamName = rorAuditDataStream.dataStream,
+      documentId = event.requestContext.id,
+      jsonRecord = serializedEvent.toString
+    )
+  }
+
+  override def close(): Task[Unit] = Task.delay(auditSinkService.close())
+
+}
+
+object EsDataStreamBasedAuditSink {
+  def create(serializer: AuditLogSerializer,
+             rorAuditDataStream: RorAuditDataStream,
+             auditSinkService: DataStreamBasedAuditSinkService): Task[EsDataStreamBasedAuditSink] = {
+    auditSinkService
+      .dataStreamCreator
+      .createIfNotExists(rorAuditDataStream)
+      .map((_: Unit) => new EsDataStreamBasedAuditSink(serializer, rorAuditDataStream, auditSinkService))
+  }
+}