ms/blueprintsprocessor/modules/commons/atomix-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/atomix/utils/AtomixLibUtils.kt

   1 /*
   2  * Copyright © 2018-2019 AT&T Intellectual Property.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 package org.onap.ccsdk.cds.blueprintsprocessor.atomix.utils
  18
  19 import com.fasterxml.jackson.databind.JsonNode
  20 import com.fasterxml.jackson.databind.node.ArrayNode
  21 import com.fasterxml.jackson.databind.node.MissingNode
  22 import com.fasterxml.jackson.databind.node.NullNode
  23 import com.fasterxml.jackson.databind.node.ObjectNode
  24 import io.atomix.core.Atomix
  25 import io.atomix.core.map.DistributedMap
  26 import io.atomix.protocols.backup.MultiPrimaryProtocol
  27 import io.atomix.protocols.backup.partition.PrimaryBackupPartitionGroup
  28 import io.atomix.protocols.raft.partition.RaftPartitionGroup
  29 import io.atomix.utils.net.Address
  30 import org.jsoup.nodes.TextNode
  31 import org.onap.ccsdk.cds.blueprintsprocessor.core.service.ClusterInfo
  32 import org.onap.ccsdk.cds.controllerblueprints.core.normalizedFile
  33
  34 object AtomixLibUtils {
  35
  36     fun configAtomix(filePath: String): Atomix {
  37         val configFile = normalizedFile(filePath)
  38         return Atomix.builder(configFile.absolutePath).build()
  39     }
  40
  41     fun defaultMulticastAtomix(clusterInfo: ClusterInfo): Atomix {
  42
  43         val nodeId = clusterInfo.nodeId
  44
  45         val raftPartitionGroup = RaftPartitionGroup.builder("system")
  46             .withNumPartitions(7)
  47             .withMembers(clusterInfo.clusterMembers)
  48             .withDataDirectory(normalizedFile("${clusterInfo.storagePath}/data-$nodeId"))
  49             .build()
  50
  51         val primaryBackupGroup =
  52             PrimaryBackupPartitionGroup.builder("data")
  53                 .withNumPartitions(31)
  54                 .build()
  55
  56         return Atomix.builder()
  57             .withMemberId(nodeId)
  58             .withAddress(Address.from(clusterInfo.nodeAddress))
  59             .withManagementGroup(raftPartitionGroup)
  60             .withPartitionGroups(primaryBackupGroup)
  61             .withMulticastEnabled()
  62             .build()
  63     }
  64
  65     fun <T> distributedMapStore(atomix: Atomix, storeName: String): DistributedMap<String, T> {
  66         check(atomix.isRunning) { "Cluster is not running, couldn't create distributed store($storeName)" }
  67
  68         val protocol = MultiPrimaryProtocol.builder()
  69             .withBackups(2)
  70             .build()
  71
  72         return atomix.mapBuilder<String, T>(storeName)
  73             .withProtocol(protocol)
  74             .withCacheEnabled()
  75             .withValueType(JsonNode::class.java)
  76             .withExtraTypes(
  77                 JsonNode::class.java, TextNode::class.java, ObjectNode::class.java,
  78                 ArrayNode::class.java, NullNode::class.java, MissingNode::class.java
  79             )
  80             .build()
  81     }
  82 }