-
Notifications
You must be signed in to change notification settings - Fork 17
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #42 from ing-bank/feature/events-emitter
Add possiblity to emit AWS like events based on user requests
- Loading branch information
Showing
18 changed files
with
346 additions
and
33 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
60 changes: 60 additions & 0 deletions
60
src/it/scala/com/ing/wbaa/airlock/proxy/provider/MessageProviderKafkaItTest.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| package com.ing.wbaa.airlock.proxy.provider | ||
|
|
||
| import java.net.InetAddress | ||
|
|
||
| import akka.actor.ActorSystem | ||
| import akka.http.scaladsl.model.{HttpMethods, RemoteAddress} | ||
| import akka.stream.ActorMaterializer | ||
| import com.ing.wbaa.airlock.proxy.config.KafkaSettings | ||
| import com.ing.wbaa.airlock.proxy.data.{AwsAccessKey, AwsRequestCredential, Read, S3Request} | ||
| import net.manub.embeddedkafka.{EmbeddedKafka, EmbeddedKafkaConfig} | ||
| import org.scalatest.{DiagrammedAssertions, WordSpecLike} | ||
| import org.scalatest.RecoverMethods._ | ||
|
|
||
| import scala.concurrent.ExecutionContext | ||
|
|
||
| class MessageProviderKafkaItTest extends WordSpecLike with DiagrammedAssertions with EmbeddedKafka with MessageProviderKafka { | ||
|
|
||
| implicit val testSystem: ActorSystem = ActorSystem("kafkaTest") | ||
|
|
||
| override implicit val kafkaSettings: KafkaSettings = KafkaSettings(testSystem) | ||
|
|
||
| override implicit val materializer: ActorMaterializer = ActorMaterializer() | ||
|
|
||
| override implicit val executionContext: ExecutionContext = testSystem.dispatcher | ||
|
|
||
| val s3Request = S3Request(AwsRequestCredential(AwsAccessKey("a"), None), Some("demobucket"), Some("s3object"), Read) | ||
| val remoteClientIP = RemoteAddress(InetAddress.getByName("127.0.0.1")) | ||
|
|
||
| "KafkaMessageProvider" should { | ||
| "Send message to correct topic with Put or Post" in { | ||
| implicit val config = EmbeddedKafkaConfig(kafkaPort = 9092) | ||
|
|
||
| withRunningKafka { | ||
| val createEventsTopic = "create_events" | ||
| createCustomTopic(createEventsTopic) | ||
|
|
||
| emitEvent(s3Request, HttpMethods.PUT, "testUser", remoteClientIP) | ||
| val result = consumeFirstStringMessageFrom(createEventsTopic) | ||
| assert(result.contains("s3:ObjectCreated:PUT")) | ||
| } | ||
| } | ||
|
|
||
| "Send message to correct topic with Delete" in { | ||
| implicit val config = EmbeddedKafkaConfig(kafkaPort = 9092) | ||
|
|
||
| withRunningKafka { | ||
| val deleteEventsTopic = "delete_events" | ||
| createCustomTopic(deleteEventsTopic) | ||
|
|
||
| emitEvent(s3Request, HttpMethods.DELETE, "testUser", remoteClientIP) | ||
| assert(consumeFirstStringMessageFrom(deleteEventsTopic).contains("s3:ObjectRemoved:DELETE")) | ||
| } | ||
| } | ||
|
|
||
| "fail on incomplete data" in { | ||
| recoverToSucceededIf[Exception](emitEvent(s3Request.copy(s3Object = None), HttpMethods.PUT, "testUser", remoteClientIP)) | ||
| } | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
37 changes: 37 additions & 0 deletions
37
src/main/scala/com/ing/wbaa/airlock/proxy/api/PostRequestActions.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,37 @@ | ||
| package com.ing.wbaa.airlock.proxy.api | ||
|
|
||
| import akka.Done | ||
| import akka.http.scaladsl.model._ | ||
| import com.ing.wbaa.airlock.proxy.config.{ AtlasSettings, KafkaSettings } | ||
| import com.ing.wbaa.airlock.proxy.data.{ LineageResponse, S3Request, User } | ||
|
|
||
| import scala.concurrent.Future | ||
|
|
||
| trait PostRequestActions { | ||
| protected[this] def atlasSettings: AtlasSettings | ||
|
|
||
| protected[this] def kafkaSettings: KafkaSettings | ||
|
|
||
| protected[this] def createLineageFromRequest(httpRequest: HttpRequest, userSTS: User, clientIPAddress: RemoteAddress): Future[LineageResponse] | ||
|
|
||
| protected[this] def emitEvent(s3Request: S3Request, method: HttpMethod, principalId: String, clientIPAddress: RemoteAddress): Future[Done] | ||
|
|
||
| def createAtlasLineage(response: HttpResponse, httpRequest: HttpRequest, userSTS: User, clientIPAddress: RemoteAddress): Unit = | ||
| if (atlasSettings.atlasEnabled && (response.status == StatusCodes.OK || response.status == StatusCodes.NoContent)) | ||
| // delete on AWS response 204 | ||
| createLineageFromRequest(httpRequest, userSTS, clientIPAddress) | ||
|
|
||
| protected[this] def createBucketNotification(response: HttpResponse, httpRequest: HttpRequest, s3Request: S3Request, | ||
| userSTS: User, clientIPAddress: RemoteAddress): Future[Done] = | ||
| httpRequest.method match { | ||
| case HttpMethods.POST | HttpMethods.PUT | HttpMethods.DELETE if kafkaSettings.kafkaEnabled && (response.status == StatusCodes.OK || response.status == StatusCodes.NoContent) => | ||
| emitEvent(s3Request, httpRequest.method, userSTS.userName.value, clientIPAddress) | ||
| case _ => Future.successful(Done) | ||
| } | ||
|
|
||
| protected[this] def handlePostRequestActions(response: HttpResponse, httpRequest: HttpRequest, s3Request: S3Request, userSTS: User, clientIPAddress: RemoteAddress): Unit = { | ||
| createAtlasLineage(response, httpRequest, userSTS, clientIPAddress) | ||
| createBucketNotification(response, httpRequest, s3Request, userSTS, clientIPAddress) | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
20 changes: 20 additions & 0 deletions
20
src/main/scala/com/ing/wbaa/airlock/proxy/config/KafkaSettings.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,20 @@ | ||
| package com.ing.wbaa.airlock.proxy.config | ||
|
|
||
| import akka.actor.{ ExtendedActorSystem, Extension, ExtensionId, ExtensionIdProvider } | ||
| import com.typesafe.config.Config | ||
|
|
||
| class KafkaSettings(config: Config) extends Extension { | ||
| val kafkaEnabled: Boolean = config.getBoolean("kafka.producer.enabled") | ||
| val bootstrapServers: String = config.getString("kafka.producer.bootstrapServers") | ||
| val createEventsTopic: String = config.getString("kafka.producer.createTopic") | ||
| val deleteEventsTopic: String = config.getString("kafka.producer.deleteTopic") | ||
| val retries: String = config.getString("kafka.producer.retries") | ||
| val retriesBackOff: String = config.getString("kafka.producer.backoff") | ||
| val retriesBackOffMax: String = config.getString("kafka.producer.backoffMax") | ||
| val kafkaConfig = config.getConfig("kafka.producer") | ||
| } | ||
|
|
||
| object KafkaSettings extends ExtensionId[KafkaSettings] with ExtensionIdProvider { | ||
| override def createExtension(system: ExtendedActorSystem): KafkaSettings = new KafkaSettings(system.settings.config) | ||
| override def lookup(): ExtensionId[KafkaSettings] = KafkaSettings | ||
| } |
Oops, something went wrong.