Merge branch 'develop' of https://github.com/bwsw/cs-vault-server int…

…o CSVS-55 # Conflicts: # src/main/scala/com/bwsw/cloudstack/vault/server/cloudstack/CloudStackService.scala # src/main/scala/com/bwsw/cloudstack/vault/server/controllers/CloudStackVaultController.scala # src/test/scala/com/bwsw/cloudstack/vault/server/cloudstack/CloudStackServiceTestSuite.scala # src/test/scala/com/bwsw/cloudstack/vault/server/cloudstack/util/CloudStackTaskCreatorTestSuite.scala # src/test/scala/com/bwsw/cloudstack/vault/server/controllers/CloudStackVaultControllerTestSuite.scala
bwsw · Oct 16, 2017 · 1f8042e · 1f8042e
2 parents d8e760d + e776b40
commit 1f8042e
Show file tree

Hide file tree

Showing 41 changed files with 347 additions and 343 deletions.
diff --git a/README.md b/README.md
@@ -15,20 +15,20 @@ Quick start
 To start a server in Docker container you should:
 
 1. Provide a file 'variables.env' containing the following required variables:
-    * `TOKEN_PERIOD` - lifetime of vault token in days (3562 days by default)
-    * `ACCOUNTS_VAULT_BASIC_PATH` - path to cloudstack accounts' secrets in vault ("secret/cs/accounts/" by default). Last "/" is required
-    * `VM_VAULT_BASIC_PATH` - path to vms' secrets in vault ("secret/cs/vms/" by default). Last "/" is required
     * `TAG_NAME_PREFIX` - prefix for token name tag
-    * `KAFKA_SERVER_LIST` - list of kafka urls
+    * `KAFKA_ENDPOINTS` - list of kafka endpoints
     * `KAFKA_TOPIC` - kafka topic containing cloudstack events
-    * `ZOOKEEPER_URL` - zookeeper url
+    * `ZOOKEEPER_ENDPOINTS` - list of zookeeper endpoints, which is separated by comma
     * `ZOOKEEPER_RETRY_DELAY` - a delay between unsuccessful connection attempt to zookeeper and repeated attempt
-    * `ZOOKEEPER_CS_VAULT_ROOT_NODE` - a root node for keeping application data in zookeeper ("/cs_vault_server" by default)
-    * `ZOOKEEPER_CS_VAULT_MASTER_LATCH_NODE` - a node for master latch keeping ("/cs_vault_server_latch" by default)
-    * `VAULT_URL` - vault url
+    * `ZOOKEEPER_ROOT_NODE` - a root node for keeping application data in zookeeper ("/cs_vault_server" by default)
+    * `ZOOKEEPER_MASTER_LATCH_NODE` - a node for master latch keeping ("/cs_vault_server_latch" by default)
+    * `VAULT_ENDPOINT` - vault endpoint
     * `VAULT_ROOT_TOKEN` - root token providing an access to a vault server
     * `VAULT_RETRY_DELAY` -  a delay between unsuccessful connection attempt to vault and repeated attempt
-    * `CS_API_URL_LIST` - list of cloudstack urls
+    * `VAULT_TOKEN_PERIOD` - lifetime of vault token in days (3562 days by default)
+    * `VAULT_ACCOUNTS_BASIC_PATH` - path to cloudstack accounts' secrets in vault ("secret/cs/accounts/" by default). Last "/" is required
+    * `VAULT_VMS_BASIC_PATH` - path to vms' secrets in vault ("secret/cs/vms/" by default). Last "/" is required
+    * `CS_ENDPOINTS` - list of cloudstack endpoints, which is separated by comma
     * `CS_API_KEY` - api key providing an access to a cloudstack server
     * `CS_SECRET_KEY` - secret key providing an access to a cloudstack server
     * `CS_RETRY_DELAY` - a delay between unsuccessful connection attempt to cloudstack and repeated attempt

diff --git a/build.sbt b/build.sbt
@@ -7,7 +7,6 @@ scalaVersion := "2.12.2"
 libraryDependencies += "br.com.autonomiccs" % "apache-cloudstack-java-client" % "1.0.5"
 libraryDependencies += "com.bettercloud" % "vault-java-driver" % "3.0.0"
 libraryDependencies += "com.typesafe" % "config" % "1.3.0"
-libraryDependencies += "com.typesafe.scala-logging" %% "scala-logging" % "3.5.0"
 libraryDependencies += "org.slf4j" % "slf4j-api" % "1.7.25"
 libraryDependencies += ("org.apache.kafka" % "kafka_2.12" % "0.10.1.1")
   .exclude("org.slf4j", "slf4j-api")

diff --git a/src/main/resources/application.conf b/src/main/resources/application.conf
@@ -1,36 +1,36 @@
 app {
-  tokenPeriod = "3562" //10 years
-  tokenPeriod = ${?TOKEN_PERIOD}
-  accountsVaultBasicPath = "secret/cs/accounts/"
-  accountsVaultBasicPath = ${?ACCOUNTS_VAULT_BASIC_PATH}
-  vmsVaultBasicPath = "secret/cs/vms/"
-  vmsVaultBasicPath = ${?VM_VAULT_BASIC_PATH}
   tagNamePrefix = ""
   tagNamePrefix = ${?TAG_NAME_PREFIX}
 
   kafka {
-    serverList = ${?KAFKA_SERVER_LIST}
+    endpoints = ${?KAFKA_ENDPOINTS}
     topic = "cs"
     topic = ${?KAFKA_TOPIC}
   }
 
   zookeeper {
-    url = ${?ZOOKEEPER_URL}
+    endpoints = ${?ZOOKEEPER_ENDPOINTS}
     retryDelay = ${?ZOOKEEPER_RETRY_DELAY}
-    zooKeeperRootNode = "/cs_vault_server"
-    zooKeeperRootNode = ${?ZOOKEEPER_CS_VAULT_ROOT_NODE}
+    rootNode = "/cs_vault_server"
+    rootNode = ${?ZOOKEEPER_ROOT_NODE}
     masterLatchNode = "/cs_vault_server_latch"
-    masterLatchNode = ${?ZOOKEEPER_CS_VAULT_MASTER_LATCH_NODE}
+    masterLatchNode = ${?ZOOKEEPER_MASTER_LATCH_NODE}
   }
 
   vault {
-    url = ${?VAULT_URL}
+    endpoint = ${?VAULT_ENDPOINT}
     rootToken = ${?VAULT_ROOT_TOKEN}
     retryDelay = ${?VAULT_RETRY_DELAY}
+    tokenPeriod = "3562" //10 years
+    tokenPeriod = ${?TOKEN_PERIOD}
+    accountsBasicPath = "secret/cs/accounts/"
+    accountsBasicPath = ${?ACCOUNTS_VAULT_BASIC_PATH}
+    vmsBasicPath = "secret/cs/vms/"
+    vmsBasicPath = ${?VM_VAULT_BASIC_PATH}
   }
 
   cloudStack {
-    apiUrlList = ${?CS_API_URL_LIST}
+    endpoints = ${?CS_ENDPOINTS}
     apiKey = ${?CS_API_KEY}
     secretKey = ${?CS_SECRET_KEY}
     retryDelay = ${?CS_RETRY_DELAY}

diff --git a/src/main/scala/com/bwsw/cloudstack/vault/server/Launcher.scala b/src/main/scala/com/bwsw/cloudstack/vault/server/Launcher.scala
@@ -21,18 +21,17 @@ package com.bwsw.cloudstack.vault.server
 import com.bwsw.cloudstack.vault.server.common.{ConfigLoader, LeaderLatch}
 import com.bwsw.cloudstack.vault.server.kafka.ConsumerManager
 import com.bwsw.cloudstack.vault.server.util.{ApplicationConfig, ConfigLiterals, DataPath}
-import com.typesafe.scalalogging.StrictLogging
+import org.slf4j.LoggerFactory
 
 import scala.util.{Failure, Success, Try}
 
 /**
   * Class is responsible for launching application.
   * It creating all services needed to start application.
   * Also it provides support the Leader-Follower registration with help ZooKeeper + Curator
-  *
-  * @author Vladislav Medvedev
   */
-object Launcher extends StrictLogging {
+object Launcher {
+  private val logger = LoggerFactory.getLogger(this.getClass)
   private var leaderLatch: Option[LeaderLatch] = None
 
   def main(args: Array[String]): Unit = {
@@ -50,12 +49,12 @@ object Launcher extends StrictLogging {
 
   protected def start(): Unit = {
     leaderLatch = Option(createLeaderLatch(
-      ApplicationConfig.getRequiredString(ConfigLiterals.zooKeeperUrl)
+      ApplicationConfig.getRequiredString(ConfigLiterals.zooKeeperEndpoints)
     ))
 
     val consumerManagerSettings = ConsumerManager.Settings(
       ApplicationConfig.getRequiredString(ConfigLiterals.kafkaTopic),
-      ApplicationConfig.getRequiredString(ConfigLiterals.kafkaServerList)
+      ApplicationConfig.getRequiredString(ConfigLiterals.kafkaEndpoints)
     )
     val components = new Components(ConfigLoader.loadConfig())
     val consumerManager = new ConsumerManager(components.cloudStackEventHandler, consumerManagerSettings)

diff --git a/src/main/scala/com/bwsw/cloudstack/vault/server/cloudstack/CloudStackService.scala b/src/main/scala/com/bwsw/cloudstack/vault/server/cloudstack/CloudStackService.scala
@@ -22,7 +22,7 @@ import java.util.UUID
 
 import com.bwsw.cloudstack.vault.server.cloudstack.entities._
 import com.bwsw.cloudstack.vault.server.cloudstack.util.CloudStackTaskCreator
-import com.bwsw.cloudstack.vault.server.cloudstack.util.exception.{CloudStackCriticalException, CloudStackEntityDoesNotExistException}
+import com.bwsw.cloudstack.vault.server.cloudstack.util.exception.CloudStackEntityDoesNotExistException
 import com.bwsw.cloudstack.vault.server.common.JsonSerializer
 import com.bwsw.cloudstack.vault.server.util.TaskRunner
 import org.slf4j.LoggerFactory
@@ -36,15 +36,15 @@ import org.slf4j.LoggerFactory
 class CloudStackService(cloudStackTaskCreator: CloudStackTaskCreator,
                         settings: CloudStackService.Settings) {
   private val logger = LoggerFactory.getLogger(this.getClass)
-  private val jsonSerializer = new JsonSerializer(true)
+  private val jsonSerializer = new JsonSerializer(ignore = true)
 
   /**
     * Gets all tags of account's users which has "User" type.
     *
     * @param accountId id of account for gets user's tags
     *
     * @return Set with Tag
-    * @throws CloudStackCriticalException if account with specified id does not exist.
+    * @throws CloudStackEntityDoesNotExistException if account with specified id does not exist.
     */
   def getUserTagsByAccount(accountId: UUID): Set[Tag] = {
     logger.debug(s"getUserTagsByAccount(accountId: $accountId)")
@@ -65,7 +65,7 @@ class CloudStackService(cloudStackTaskCreator: CloudStackTaskCreator,
     * @param userId id of user for gets user's tags
     *
     * @return Set with Tag
-    * @throws CloudStackCriticalException if user with specified id does not exist.
+    * @throws CloudStackEntityDoesNotExistException if user with specified id does not exist.
     */
   def getUserTags(userId: UUID): Set[Tag] = {
     logger.debug(s"getUserTags(userId: $userId)")
@@ -83,7 +83,7 @@ class CloudStackService(cloudStackTaskCreator: CloudStackTaskCreator,
     * @param vmId id of virtual mashine for gets user's tags
     *
     * @return Set with Tag
-    * @throws CloudStackCriticalException if virtual machine with specified id does not exist.
+    * @throws CloudStackEntityDoesNotExistException if virtual machine with specified id does not exist.
     */
   def getVmTags(vmId: UUID): Set[Tag] = {
     logger.debug(s"getVmTags(vmId: $vmId)")
@@ -102,22 +102,22 @@ class CloudStackService(cloudStackTaskCreator: CloudStackTaskCreator,
     * @param vmId id of virtual machine for gets account name
     *
     * @return UUID of account which name indicate in virtual machine
-    * @throws CloudStackCriticalException if virtual machine with specified id does not exist,
-    *                                     or if account with specified name in virtual machine does not exist.
+    * @throws CloudStackEntityDoesNotExistException if virtual machine with specified id does not exist,
+    *                                               or if account with specified name in virtual machine does not exist.
     */
   def getVmOwnerAccount(vmId: UUID): UUID = {
     logger.debug(s"getVmOwnerAccount(vmId: $vmId)")
 
     val accountName = jsonSerializer.deserialize[VirtualMachinesResponse](
       getEntityJson(vmId.toString, cloudStackTaskCreator.idParameter, Command.ListVirtualMachines)
     ).virtualMashineList.virtualMashines.getOrElse(
-      throw new CloudStackCriticalException(new CloudStackEntityDoesNotExistException(s"Virtual machine with id: $vmId does not exist"))
+      throw new CloudStackEntityDoesNotExistException(s"Virtual machine with id: $vmId does not exist")
     ).map(_.accountName).head
 
-    val accountId: UUID = jsonSerializer.deserialize[AccountResponse](
+    val accountId = jsonSerializer.deserialize[AccountResponse](
       getEntityJson(accountName, cloudStackTaskCreator.nameParameter, Command.ListAccounts)
     ).accountList.accounts.getOrElse(
-      throw new CloudStackCriticalException(new CloudStackEntityDoesNotExistException(s"The vm: $vmId does not include account with name: $accountName"))
+      throw new CloudStackEntityDoesNotExistException(s"The vm: $vmId does not include account with name: $accountName")
     ).map(_.id).head
 
     logger.debug(s"accountId was got for vm: $vmId)")
@@ -130,15 +130,15 @@ class CloudStackService(cloudStackTaskCreator: CloudStackTaskCreator,
     * @param userId id of user for gets account id
     *
     * @return UUID of account which include user with indicate id
-    * @throws CloudStackCriticalException if user with specified id does not exist.
+    * @throws CloudStackEntityDoesNotExistException if user with specified id does not exist.
     */
   def getAccountByUser(userId: UUID): UUID = {
     logger.debug(s"getAccountByUser(userId: $userId)")
 
     val accountId = jsonSerializer.deserialize[UserResponse](
       getEntityJson(userId.toString, cloudStackTaskCreator.idParameter, Command.ListUsers)
     ).userList.users.getOrElse(
-      throw new CloudStackCriticalException(new CloudStackEntityDoesNotExistException(s"User with id: $userId does not exist"))
+      throw new CloudStackEntityDoesNotExistException(s"User with id: $userId does not exist")
     ).map(_.accountid).head
 
     logger.debug(s"accountId was got for user: $userId)")
@@ -151,11 +151,11 @@ class CloudStackService(cloudStackTaskCreator: CloudStackTaskCreator,
     * @param accountId id of user for gets account id
     *
     * @return List with UUID of users which are included in account
-    * @throws CloudStackCriticalException if account with specified id does not exist.
+    * @throws CloudStackEntityDoesNotExistException if account with specified id does not exist.
     */
   def getUsersByAccount(accountId: UUID): List[UUID] = {
     logger.debug(s"getUsersByAccount(accountId: $accountId)")
-    val jsonSerializer = new JsonSerializer(true)
+    val jsonSerializer = new JsonSerializer(ignore = true)
 
     val accountResponse = getEntityJson(
       accountId.toString,
@@ -166,7 +166,7 @@ class CloudStackService(cloudStackTaskCreator: CloudStackTaskCreator,
     val allUsersIdInAccount = jsonSerializer.deserialize[AccountResponse](accountResponse)
       .accountList
       .accounts.getOrElse(
-        throw new CloudStackCriticalException(new CloudStackEntityDoesNotExistException(s"Account with id: $accountId does not exist"))
+        throw new CloudStackEntityDoesNotExistException(s"Account with id: $accountId does not exist")
       ).flatMap { x =>
         x.users.map(_.id)
       }
@@ -186,23 +186,23 @@ class CloudStackService(cloudStackTaskCreator: CloudStackTaskCreator,
     logger.debug(s"setResourceTags(resourceId: $resourceId, resourceType: $resourceType)")
     def task = cloudStackTaskCreator.createSetResourceTagsTask(resourceId, resourceType, tagSet)
 
-    TaskRunner.tryRunUntilSuccess[Unit](task, settings.cloudStackRetryDelay)
+    TaskRunner.tryRunUntilSuccess[Unit](task, settings.retryDelay)
     logger.debug(s"Tag was set to resource: $resourceId, $resourceType")
   }
 
   private def getEntityJson(parameterValue: String, parameterName: String, command: Command): String = {
     def task = cloudStackTaskCreator.createGetEntityTask(parameterValue, parameterName, command)
 
-    TaskRunner.tryRunUntilSuccess[String](task, settings.cloudStackRetryDelay)
+    TaskRunner.tryRunUntilSuccess[String](task, settings.retryDelay)
   }
 
   private def getTagsJson(resourceType: Tag.Type, resourceId: UUID): String = {
     def task = cloudStackTaskCreator.createGetTagTask(resourceType, resourceId)
 
-    TaskRunner.tryRunUntilSuccess[String](task, settings.cloudStackRetryDelay)
+    TaskRunner.tryRunUntilSuccess[String](task, settings.retryDelay)
   }
 }
 
 object CloudStackService {
-  case class Settings(cloudStackRetryDelay: Int)
+  case class Settings(retryDelay: Int)
 }
diff --git a/src/main/scala/com/bwsw/cloudstack/vault/server/cloudstack/entities/CloudStackEvent.scala b/src/main/scala/com/bwsw/cloudstack/vault/server/cloudstack/entities/CloudStackEvent.scala
@@ -21,9 +21,9 @@ package com.bwsw.cloudstack.vault.server.cloudstack.entities
 import java.util.UUID
 
 import com.fasterxml.jackson.annotation.JsonProperty
-import com.fasterxml.jackson.core.{JsonGenerator, JsonParser}
+import com.fasterxml.jackson.core.JsonParser
 import com.fasterxml.jackson.databind._
-import com.fasterxml.jackson.databind.annotation.{JsonDeserialize, JsonSerialize}
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize
 
 object CloudStackEvent {
 

diff --git a/src/main/scala/com/bwsw/cloudstack/vault/server/cloudstack/util/CloudStackEventHandler.scala b/src/main/scala/com/bwsw/cloudstack/vault/server/cloudstack/util/CloudStackEventHandler.scala
@@ -19,7 +19,7 @@
 package com.bwsw.cloudstack.vault.server.cloudstack.util
 
 import com.bwsw.cloudstack.vault.server.cloudstack.entities.CloudStackEvent
-import com.bwsw.cloudstack.vault.server.common.JsonSerializer
+import com.bwsw.cloudstack.vault.server.common.{JsonSerializer, ProcessingEventResult}
 import com.bwsw.cloudstack.vault.server.common.traits.EventHandler
 import com.bwsw.cloudstack.vault.server.controllers.CloudStackVaultController
 import org.slf4j.LoggerFactory
@@ -39,11 +39,11 @@ import scala.util.{Failure, Success, Try}
   */
 class CloudStackEventHandler(controller: CloudStackVaultController)
                             (implicit executionContext: ExecutionContext) extends EventHandler[CloudStackEvent] {
-  private val jsonSerializer = new JsonSerializer(true)
+  private val jsonSerializer = new JsonSerializer(ignore = true)
   private val logger = LoggerFactory.getLogger(this.getClass)
 
   @Override
-  def handleEventsFromRecords(records: List[String]): Set[(Future[Unit], CloudStackEvent)] = {
+  def handleEventsFromRecords(records: List[String]): Set[ProcessingEventResult[CloudStackEvent]] = {
     logger.debug(s"handleEventsFromRecords: $records")
     records.map { record =>
       Try {
@@ -61,29 +61,29 @@ class CloudStackEventHandler(controller: CloudStackVaultController)
   }
 
   @Override
-  def restartEvent(event: CloudStackEvent): (Future[Unit], CloudStackEvent) = {
+  def restartEvent(event: CloudStackEvent): ProcessingEventResult[CloudStackEvent] = {
     logger.debug(s"restartEvent: $event")
     handleEvent(event)
   }
 
-  private val handleEvent = new PartialFunction[CloudStackEvent, (Future[Unit], CloudStackEvent)] {
-    override def apply(event: CloudStackEvent): (Future[Unit], CloudStackEvent) = {
+  private val handleEvent = new PartialFunction[CloudStackEvent, ProcessingEventResult[CloudStackEvent]] {
+    override def apply(event: CloudStackEvent): ProcessingEventResult[CloudStackEvent] = {
       event.action.get match {
         case VMCreate =>
           logger.info(s"handle VMCreate event: $event")
-          (Future(controller.handleVmCreate(event.entityuuid.get)), event)
+          ProcessingEventResult(event, Future(controller.handleVmCreate(event.entityuuid.get)))
         case VMDelete =>
           logger.info(s"handle VMDelete event: $event")
-          (Future(controller.handleVmDelete(event.entityuuid.get)), event)
+          ProcessingEventResult(event, Future(controller.handleVmDelete(event.entityuuid.get)))
         case AccountCreate =>
           logger.info(s"handle AccountCreate event: $event")
-          (Future(controller.handleAccountCreate(event.entityuuid.get)), event)
+          ProcessingEventResult(event, Future(controller.handleAccountCreate(event.entityuuid.get)))
         case AccountDelete =>
           logger.info(s"handle AccountDelete event: $event")
-          (Future(controller.handleAccountDelete(event.entityuuid.get)), event)
+          ProcessingEventResult(event, Future(controller.handleAccountDelete(event.entityuuid.get)))
         case UserCreate =>
           logger.info(s"handle UserCreate event: $event")
-          (Future(controller.handleUserCreate(event.entityuuid.get)), event)
+          ProcessingEventResult(event, Future(controller.handleUserCreate(event.entityuuid.get)))
       }
     }
 
@@ -100,16 +100,4 @@ class CloudStackEventHandler(controller: CloudStackVaultController)
       }
     }
   }
-
-  def isNonFatalException(exception: Throwable): Boolean = {
-    logger.debug(s"isNonFatalException: $exception")
-    exception match {
-      case e: CriticalException =>
-        e.exception match {
-          case nonFatalException: CloudStackEntityDoesNotExistException => true
-          case _ => false
-        }
-      case _ => false
-    }
-  }
 }