2 * Copyright © 2018-2019 AT&T Intellectual Property.
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
8 * http://www.apache.org/licenses/LICENSE-2.0
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.
17 package org.onap.ccsdk.cds.blueprintsprocessor.atomix
19 import com.fasterxml.jackson.databind.JsonNode
20 import kotlinx.coroutines.Dispatchers
21 import kotlinx.coroutines.async
22 import kotlinx.coroutines.awaitAll
23 import kotlinx.coroutines.delay
24 import kotlinx.coroutines.runBlocking
25 import kotlinx.coroutines.withContext
26 import org.junit.Before
28 import org.onap.ccsdk.cds.blueprintsprocessor.atomix.service.AtomixBluePrintClusterService
29 import org.onap.ccsdk.cds.blueprintsprocessor.core.service.BluePrintClusterService
30 import org.onap.ccsdk.cds.blueprintsprocessor.core.service.ClusterInfo
31 import org.onap.ccsdk.cds.controllerblueprints.core.asJsonPrimitive
32 import org.onap.ccsdk.cds.controllerblueprints.core.deleteNBDir
33 import org.onap.ccsdk.cds.controllerblueprints.core.logger
34 import kotlin.test.assertNotNull
35 import kotlin.test.assertTrue
37 class AtomixBluePrintClusterServiceTest {
38 private val log = logger(AtomixBluePrintClusterServiceTest::class)
43 deleteNBDir("target/cluster")
47 /** Testing two cluster with distributed map store creation, This is time consuming test case, taks around 10s **/
49 fun testClusterJoin() {
51 val bluePrintClusterServiceOne =
52 createCluster(arrayListOf(5679, 5680)).toMutableList()
53 // val bluePrintClusterServiceTwo = createCluster(arrayListOf(5681, 5682), arrayListOf(5679, 5680))
54 // bluePrintClusterServiceOne.addAll(bluePrintClusterServiceTwo)
55 val bluePrintClusterService = bluePrintClusterServiceOne[0]
56 log.info("Members : ${bluePrintClusterService.allMembers()}")
57 log.info("Master(System) Members : ${bluePrintClusterService.masterMember("system")}")
58 log.info("Master(Data) Members : ${bluePrintClusterService.masterMember("data")}")
59 testDistributedStore(bluePrintClusterServiceOne)
60 testDistributedLock(bluePrintClusterServiceOne)
64 private suspend fun createCluster(
66 otherClusterPorts: List<Int>? = null
67 ): List<BluePrintClusterService> {
69 return withContext(Dispatchers.Default) {
70 val clusterMembers = ports.map { "node-$it" }.toMutableList()
71 /** Add the other cluster as members */
72 if (!otherClusterPorts.isNullOrEmpty()) {
73 val otherClusterMembers = otherClusterPorts.map { "node-$it" }.toMutableList()
74 clusterMembers.addAll(otherClusterMembers)
76 val deferred = ports.map { port ->
77 async(Dispatchers.IO) {
78 val nodeId = "node-$port"
79 log.info("********** Starting node($nodeId) on port($port)")
80 val clusterInfo = ClusterInfo(
81 id = "test-cluster", nodeId = nodeId,
82 clusterMembers = clusterMembers, nodeAddress = "localhost:$port", storagePath = "target/cluster"
84 val atomixClusterService = AtomixBluePrintClusterService()
85 atomixClusterService.startCluster(clusterInfo)
93 private suspend fun testDistributedStore(bluePrintClusterServices: List<BluePrintClusterService>) {
94 /** Test Distributed store creation */
95 repeat(2) { storeId ->
96 val store = bluePrintClusterServices[0].clusterMapStore<JsonNode>(
97 "blueprint-runtime-$storeId"
99 assertNotNull(store, "failed to get store")
100 val store1 = bluePrintClusterServices[1].clusterMapStore<JsonNode>(
101 "blueprint-runtime-$storeId"
105 log.info("Received map event : $it")
108 store["key-$storeId-$it"] = "value-$it".asJsonPrimitive()
115 private suspend fun testDistributedLock(bluePrintClusterServices: List<BluePrintClusterService>) {
116 val lockName = "sample-lock"
117 withContext(Dispatchers.IO) {
118 val deferred = async {
119 executeLock(bluePrintClusterServices[0], "first", lockName)
121 val deferred2 = async {
122 executeLock(bluePrintClusterServices[0], "second", lockName)
124 val deferred3 = async {
125 executeLock(bluePrintClusterServices[1], "third", lockName)
133 private suspend fun executeLock(
134 bluePrintClusterService: BluePrintClusterService,
138 log.info("initialising $lockId lock...")
139 val distributedLock = bluePrintClusterService.clusterLock(lockName)
140 assertNotNull(distributedLock, "failed to create distributed $lockId lock")
141 distributedLock.lock()
142 assertTrue(distributedLock.isLocked(), "failed to lock $lockId")
144 log.info("locked $lockId process for 5mSec")
147 distributedLock.unLock()
148 log.info("$lockId lock released")
150 distributedLock.close()