Remove Atomix implementation, due to Kubernetes clustering issues.
Cluster environment property changes.
Issue-ID: CCSDK-2011
Signed-off-by: Brinda Santh <bs2796@att.com>
Change-Id: I23f40c92c0adc6b3ab8690871385f78525c76433
<artifactId>health-api-common</artifactId>
</dependency>
- <!-- Libs -->
- <dependency>
- <groupId>org.onap.ccsdk.cds.blueprintsprocessor</groupId>
- <artifactId>atomix-lib</artifactId>
- </dependency>
-
<!-- Functions -->
<dependency>
<groupId>org.onap.ccsdk.cds.blueprintsprocessor.functions</groupId>
- cds-network
ports:
- "8000:8080"
- - "9111:9111"
+ - "9110:9111"
restart: always
volumes:
- target: /opt/app/onap/blueprints/deploy
CLUSTER_ENABLED: "true"
CLUSTER_ID: cds-cluster
CLUSTER_NODE_ID: cds-controller-0
- CLUSTER_MEMBERS: cds-controller-0,resource-resolution-0
- CLUSTER_STORAGE_PATH: /opt/app/onap/config/cluster
- #CLUSTER_CONFIG_FILE: /opt/app/onap/config/atomix/atomix-multicast.conf
+ CLUSTER_JOIN_AS_CLIENT: "false"
+ CLUSTER_CONFIG_FILE: /opt/app/onap/config/hazelcast/hazelcast.yaml
NATS_CLUSTER_ID: cds-cluster
APP_NAME: cds-controller
BUNDLEVERSION: 1.0.0
APP_CONFIG_HOME: /opt/app/onap/config
STICKYSELECTORKEY:
ENVCONTEXT: dev
- resource-resolution-0:
+ cds-controller-1:
depends_on:
- db
- nats
image: onap/ccsdk-blueprintsprocessor:latest
- container_name: resource-resolution-0
- hostname: resource-resolution-0
+ container_name: cds-controller-1
+ hostname: cds-controller-1
networks:
- cds-network
ports:
- "8001:8080"
+ - "9111:9111"
+ restart: always
+ volumes:
+ - target: /opt/app/onap/blueprints/deploy
+ type: volume
+ source: blueprints-deploy
+ - target: /opt/app/onap/config
+ type: volume
+ source: controller-config
+ environment:
+ # Same as hostname and container name
+ CLUSTER_ENABLED: "true"
+ CLUSTER_ID: cds-cluster
+ CLUSTER_NODE_ID: cds-controller-1
+ CLUSTER_JOIN_AS_CLIENT: "false"
+ CLUSTER_CONFIG_FILE: /opt/app/onap/config/hazelcast/hazelcast.yaml
+ NATS_CLUSTER_ID: cds-cluster
+ APP_NAME: cds-controller
+ BUNDLEVERSION: 1.0.0
+ APP_CONFIG_HOME: /opt/app/onap/config
+ STICKYSELECTORKEY:
+ ENVCONTEXT: dev
+ cds-controller-2:
+ depends_on:
+ - db
+ - nats
+ image: onap/ccsdk-blueprintsprocessor:latest
+ container_name: cds-controller-2
+ hostname: cds-controller-2
+ networks:
+ - cds-network
+ ports:
+ - "8002:8080"
- "9112:9111"
restart: always
volumes:
source: blueprints-deploy
- target: /opt/app/onap/config
type: volume
- source: resource-resolution-config
+ source: controller-config
environment:
+ # Same as hostname and container name
CLUSTER_ENABLED: "true"
CLUSTER_ID: cds-cluster
- CLUSTER_NODE_ID: resource-resolution-0
- CLUSTER_MEMBERS: cds-controller-0,resource-resolution-0
- CLUSTER_STORAGE_PATH: /opt/app/onap/config/cluster
- #CLUSTER_CONFIG_FILE: /opt/app/onap/config/atomix/atomix-multicast.conf
+ CLUSTER_NODE_ID: cds-controller-2
+ CLUSTER_JOIN_AS_CLIENT: "false"
+ CLUSTER_CONFIG_FILE: /opt/app/onap/config/hazelcast/hazelcast.yaml
NATS_CLUSTER_ID: cds-cluster
- APP_NAME: resource-resolution
+ APP_NAME: cds-controller
BUNDLEVERSION: 1.0.0
APP_CONFIG_HOME: /opt/app/onap/config
STICKYSELECTORKEY:
environment:
CLUSTER_ID: cds-cluster
CLUSTER_NODE_ID: py-executor-0
- CLUSTER_MEMBERS: cds-controller-0,resource-resolution-0,py-executor-0
+ CLUSTER_JOIN_AS_CLIENT: "false"
+ CLUSTER_CONFIG_FILE: /opt/app/onap/config/hazelcast/hazelcast.yaml
NATS_CLUSTER_ID: cds-cluster
APP_NAME: py-executor
BUNDLEVERSION: 1.0.0
driver: local
driver_opts:
type: none
- device: /opt/app/cds/nats/nats-0/store
+ device: /opt/app/cds/nats/store
o: bind
blueprints-deploy:
driver: local
driver: local
driver_opts:
type: none
- device: /opt/app/cds/cds-controller/cds-controller-0/config
- o: bind
- resource-resolution-config:
- driver: local
- driver_opts:
- type: none
- device: /opt/app/cds/resource-resolution/resource-resolution-0/config
+ device: /opt/app/cds/cds-controller/config
o: bind
networks:
<outputDirectory>opt/app/onap/config</outputDirectory>
<useDefaultExcludes>true</useDefaultExcludes>
</fileSet>
+ <fileSet>
+ <directory>${project.basedir}/src/main/resources/hazelcast</directory>
+ <outputDirectory>opt/app/onap/config/hazelcast</outputDirectory>
+ <useDefaultExcludes>true</useDefaultExcludes>
+ </fileSet>
<fileSet>
<directory>${project.basedir}/src/main/resources/certs</directory>
<outputDirectory>opt/app/onap/config/certs</outputDirectory>
import org.onap.ccsdk.cds.blueprintsprocessor.core.service.BluePrintClusterService
import org.onap.ccsdk.cds.blueprintsprocessor.core.service.ClusterInfo
import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants
-import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
import org.onap.ccsdk.cds.controllerblueprints.core.logger
-import org.onap.ccsdk.cds.controllerblueprints.core.splitCommaAsList
import org.onap.ccsdk.cds.controllerblueprints.core.utils.ClusterUtils
import org.springframework.boot.context.event.ApplicationReadyEvent
import org.springframework.context.event.EventListener
import org.springframework.stereotype.Component
import java.time.Duration
+import java.util.Properties
import javax.annotation.PreDestroy
/**
* 2. Container names should end with sequence number.
* Blueprintprocessor example be : cds-controller-1, cds-controller-2, cds-controller-3
* ResourceResolution example be : resource-resolution-1, resource-resolution-2, resource-resolution-3
- * 3. Each contained, should have environment properties CLUSTER_ID, CLUSTER_NODE_ID, CLUSTER_NODE_ADDRESS,
- * CLUSTER_MEMBERS, CLUSTER_STORAGE_PATH
+ * 3. Each contained, should have environment properties CLUSTER_ID, CLUSTER_NODE_ID, CLUSTER_JOIN_AS_CLIENT,
+ * CLUSTER_CONFIG_FILE
* Example values :
* CLUSTER_ID: cds-cluster
* CLUSTER_NODE_ID: cds-controller-2
- * CLUSTER_NODE_ADDRESS: cds-controller-2
- * CLUSTER_MEMBERS: cds-controller-1,cds-controller-2,cds-controller-3,resource-resolution-1,resource-resolution-2,resource-resolution-3
- * CLUSTER_STORAGE_PATH: /opt/app/onap/config/cluster
- * CLUSTER_CONFIG_FILE: /opt/app/onap/config/atomix/atomix-multicast.conf
+ * CLUSTER_JOIN_AS_CLIENT: "true" or "false"
+ * CLUSTER_CONFIG_FILE: <Config location>
* 4. Cluster will be enabled only all the above properties present in the environments.
* if CLUSTER_ENABLED is present, then it will try to create cluster.
*/
val clusterId = ClusterUtils.clusterId()
val nodeId = ClusterUtils.clusterNodeId()
- val nodeAddress = ClusterUtils.clusterNodeAddress()
- val clusterMembers = System.getenv(BluePrintConstants.PROPERTY_CLUSTER_MEMBERS)
- ?: throw BluePrintProcessorException("couldn't get environment variable ${BluePrintConstants.PROPERTY_CLUSTER_MEMBERS}")
-
- val clusterMemberList = clusterMembers.splitCommaAsList()
-
- val clusterStorage = System.getenv(BluePrintConstants.PROPERTY_CLUSTER_STORAGE_PATH)
- ?: throw BluePrintProcessorException("couldn't get environment variable ${BluePrintConstants.PROPERTY_CLUSTER_STORAGE_PATH}")
+ val joinAsClient =
+ (System.getenv(BluePrintConstants.PROPERTY_CLUSTER_JOIN_AS_CLIENT) ?: "false").toBoolean()
val clusterConfigFile = System.getenv(BluePrintConstants.PROPERTY_CLUSTER_CONFIG_FILE)
+ val properties = Properties()
+ properties["hazelcast.logging.type"] = "slf4j"
+
val clusterInfo = ClusterInfo(
id = clusterId, nodeId = nodeId,
- clusterMembers = clusterMemberList, nodeAddress = nodeAddress,
- storagePath = clusterStorage,
- configFile = clusterConfigFile
+ joinAsClient = joinAsClient,
+ configFile = clusterConfigFile,
+ properties = properties
)
bluePrintClusterService.startCluster(clusterInfo)
} else {
+++ /dev/null
-cluster {
- # Configure the cluster node information.
- node {
- id: ${CLUSTER_NODE_ID}
- address: ${CLUSTER_NODE_ADDRESS}
- }
- # Configure the node discovery protocol.
- discovery {
- type: bootstrap
- nodes.1 {
- id: cds-controller-1
- address: "cds-controller-1:5679"
- }
- nodes.2 {
- id: resource-reolution-1
- address: "resource-reolution-1:5679"
- }
- }
-}
-# Configure the system management group.
-managementGroup {
- type: raft
- name: system
- partitions: 1
- members: [${CLUSTER_MEMBERS}]
- storage {
- directory: ${CLUSTER_STORAGE_PATH}/data-${CLUSTER_NODE_ID}
- level: DISK
- }
-}
-# Configure a Raft partition group.
-partitionGroups.data {
- type: primary-backup
- partitions: 7
-}
+++ /dev/null
-cluster {
- # Configure the cluster node information.
- node {
- id: ${CLUSTER_NODE_ID}
- address: ${CLUSTER_NODE_ADDRESS}
- }
- # Configure the node discovery protocol.
- discovery {
- type: multicast
- }
- multicast: {
- enabled: true
- port: 54321
- }
- # Configure the SWIM membership protocol.
- protocol {
- type: swim
- broadcastUpdates: true
- gossipInterval: 500ms
- probeInterval: 2s
- suspectProbes: 2
- }
-}
-# Configure the system management group.
-managementGroup {
- type: raft
- name: system
- partitions: 1
- members: [${CLUSTER_MEMBERS}]
- storage {
- directory: ${CLUSTER_STORAGE_PATH}/data-${CLUSTER_NODE_ID}
- level: DISK
- }
-}
-
-# Configure a Raft partition group.
-partitionGroups.data {
- type: primary-backup
- partitions: 7
-}
--- /dev/null
+hazelcast-client:
+ cluster-name: ${CLUSTER_ID}
+ instance-name: ${CLUSTER_NODE_ID}
+
+ network:
+ cluster-members:
+ - 127.0.0.1:5701
+# kubernetes:
+# enabled: true
+# namespace: MY-KUBERNETES-NAMESPACE
+# service-name: MY-SERVICE-NAME
+# service-label-name: MY-SERVICE-LABEL-NAME
+# service-label-value: MY-SERVICE-LABEL-VALUE
--- /dev/null
+hazelcast:
+ cluster-name: ${CLUSTER_ID}
+ instance-name: ${CLUSTER_NODE_ID}
+ lite-member:
+ enabled: false
+ cp-subsystem:
+ cp-member-count: 3
+ group-size: 3
+# network:
+# join:
+# multicast:
+# enabled: false
+# kubernetes:
+# enabled: true
+# namespace: MY-KUBERNETES-NAMESPACE
+# service-name: MY-SERVICE-NAME
+# service-label-name: MY-SERVICE-LABEL-NAME
+# service-label-value: MY-SERVICE-LABEL-VALUE
\ No newline at end of file
<logger name="org.springframework" level="info"/>
<logger name="org.springframework.web" level="info"/>
<logger name="org.hibernate" level="error"/>
- <logger name="io.atomix" level="warn"/>
+ <logger name="com.hazelcast" level="error"/>
<logger name="org.onap.ccsdk.cds" level="info"/>
<root level="info">
<description>Blueprints Processor Function - Message Prioritization</description>
<dependencies>
- <dependency>
- <groupId>org.onap.ccsdk.cds.blueprintsprocessor</groupId>
- <artifactId>atomix-lib</artifactId>
- </dependency>
<dependency>
<groupId>org.onap.ccsdk.cds.blueprintsprocessor</groupId>
<artifactId>message-lib</artifactId>
package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.utils
import org.apache.kafka.streams.processor.ProcessorSupplier
-import org.onap.ccsdk.cds.blueprintsprocessor.atomix.optionalClusterService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.cluster.optionalClusterService
import org.onap.ccsdk.cds.blueprintsprocessor.core.service.ClusterLock
import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization
import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.kafka.AbstractMessagePrioritizeProcessor
const val PROPERTY_CLUSTER_ID = "CLUSTER_ID"
const val PROPERTY_CLUSTER_NODE_ID = "CLUSTER_NODE_ID"
const val PROPERTY_CLUSTER_NODE_ADDRESS = "CLUSTER_NODE_ADDRESS"
- const val PROPERTY_CLUSTER_MEMBERS = "CLUSTER_MEMBERS"
- const val PROPERTY_CLUSTER_STORAGE_PATH = "CLUSTER_STORAGE_PATH"
+ const val PROPERTY_CLUSTER_JOIN_AS_CLIENT = "CLUSTER_JOIN_AS_CLIENT"
const val PROPERTY_CLUSTER_CONFIG_FILE = "CLUSTER_CONFIG_FILE"
}
+++ /dev/null
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- ~ Copyright © 2018-2019 AT&T Intellectual Property.
- ~
- ~ Licensed under the Apache License, Version 2.0 (the "License");
- ~ you may not use this file except in compliance with the License.
- ~ You may obtain a copy of the License at
- ~
- ~ http://www.apache.org/licenses/LICENSE-2.0
- ~
- ~ Unless required by applicable law or agreed to in writing, software
- ~ distributed under the License is distributed on an "AS IS" BASIS,
- ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- ~ See the License for the specific language governing permissions and
- ~ limitations under the License.
- -->
-
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
-
- <parent>
- <groupId>org.onap.ccsdk.cds.blueprintsprocessor</groupId>
- <artifactId>commons</artifactId>
- <version>0.7.0-SNAPSHOT</version>
- </parent>
-
- <artifactId>atomix-lib</artifactId>
- <packaging>jar</packaging>
-
- <name>Blueprints Processor Atomix Lib</name>
- <description>Blueprints Processor Atomix Lib</description>
-
- <dependencies>
- <dependency>
- <groupId>io.atomix</groupId>
- <artifactId>atomix</artifactId>
- </dependency>
- <dependency>
- <groupId>io.atomix</groupId>
- <artifactId>atomix-raft</artifactId>
- </dependency>
- <dependency>
- <groupId>io.atomix</groupId>
- <artifactId>atomix-primary-backup</artifactId>
- </dependency>
- <dependency>
- <groupId>io.atomix</groupId>
- <artifactId>atomix-gossip</artifactId>
- </dependency>
-
- <dependency>
- <groupId>org.onap.ccsdk.cds.blueprintsprocessor</groupId>
- <artifactId>db-lib</artifactId>
- </dependency>
- </dependencies>
-</project>
+++ /dev/null
-/*
- * Copyright © 2018-2019 AT&T Intellectual Property.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.onap.ccsdk.cds.blueprintsprocessor.atomix
-
-import com.fasterxml.jackson.databind.JsonNode
-import io.atomix.core.map.DistributedMap
-import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
-
-fun <T : Map<String, JsonNode>> T.toDistributedMap(): DistributedMap<String, JsonNode> {
- return if (this != null && this is DistributedMap<*, *>) this as DistributedMap<String, JsonNode>
- else throw BluePrintProcessorException("map is not of type DistributedMap")
-}
+++ /dev/null
-/*
- * Copyright © 2018-2019 AT&T Intellectual Property.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.onap.ccsdk.cds.blueprintsprocessor.atomix.service
-
-import io.atomix.cluster.ClusterMembershipEvent
-import io.atomix.core.Atomix
-import io.atomix.core.lock.DistributedLock
-import kotlinx.coroutines.delay
-import org.onap.ccsdk.cds.blueprintsprocessor.atomix.utils.AtomixLibUtils
-import org.onap.ccsdk.cds.blueprintsprocessor.core.service.BluePrintClusterService
-import org.onap.ccsdk.cds.blueprintsprocessor.core.service.ClusterInfo
-import org.onap.ccsdk.cds.blueprintsprocessor.core.service.ClusterLock
-import org.onap.ccsdk.cds.blueprintsprocessor.core.service.ClusterMember
-import org.onap.ccsdk.cds.controllerblueprints.core.logger
-import org.springframework.stereotype.Service
-import java.time.Duration
-import java.util.concurrent.CompletableFuture
-
-@Service
-open class AtomixBluePrintClusterService : BluePrintClusterService {
-
- private val log = logger(AtomixBluePrintClusterService::class)
-
- lateinit var atomix: Atomix
-
- override suspend fun startCluster(clusterInfo: ClusterInfo) {
- log.info(
- "Cluster(${clusterInfo.id}) node(${clusterInfo.nodeId}), node address(${clusterInfo.nodeAddress}) " +
- "starting with members(${clusterInfo.clusterMembers})"
- )
-
- /** Create Atomix cluster either from config file or default multi-cast cluster*/
- atomix = if (!clusterInfo.configFile.isNullOrEmpty()) {
- AtomixLibUtils.configAtomix(clusterInfo.configFile!!)
- } else {
- AtomixLibUtils.defaultMulticastAtomix(clusterInfo)
- }
-
- /** Listen for the member chaneg events */
- atomix.membershipService.addListener { membershipEvent ->
- when (membershipEvent.type()) {
- ClusterMembershipEvent.Type.MEMBER_ADDED -> log.info("Member Added : ${membershipEvent.subject()}")
- ClusterMembershipEvent.Type.MEMBER_REMOVED -> log.info("Member Removed: ${membershipEvent.subject()}")
- ClusterMembershipEvent.Type.REACHABILITY_CHANGED -> log.info("Reachability Changed : ${membershipEvent.subject()}")
- ClusterMembershipEvent.Type.METADATA_CHANGED -> log.info("Changed : ${membershipEvent.subject()}")
- else -> log.info("Member event unknown")
- }
- }
- /** Start and Join the Cluster */
- atomix.start().join()
- log.info(
- "Cluster(${clusterInfo.id}) node(${clusterInfo.nodeId}), node address(${clusterInfo.nodeAddress}) " +
- "created successfully...."
- )
-
- /** Receive ping from network */
- val pingHandler = { message: String ->
- log.info("####### ping message received : $message")
- CompletableFuture.completedFuture(message)
- }
- atomix.communicationService.subscribe("ping", pingHandler)
-
- /** Ping the network */
- atomix.communicationService.broadcast(
- "ping",
- "ping from node(${clusterInfo.nodeId})"
- )
- }
-
- override fun clusterJoined(): Boolean {
- return atomix.isRunning
- }
-
- override suspend fun masterMember(partitionGroup: String): ClusterMember {
- check(::atomix.isInitialized) { "failed to start and join cluster" }
- check(atomix.isRunning) { "cluster is not running" }
- val masterId = atomix.partitionService
- .getPartitionGroup(partitionGroup)
- .getPartition("1").primary()
- val masterMember = atomix.membershipService.getMember(masterId)
- return ClusterMember(
- id = masterMember.id().id(),
- memberAddress = masterMember.address().toString()
- )
- }
-
- override suspend fun allMembers(): Set<ClusterMember> {
- check(::atomix.isInitialized) { "failed to start and join cluster" }
- check(atomix.isRunning) { "cluster is not running" }
-
- return atomix.membershipService.members.map {
- ClusterMember(
- id = it.id().id(),
- memberAddress = it.address().toString()
- )
- }.toSet()
- }
-
- override suspend fun clusterMembersForPrefix(memberPrefix: String): Set<ClusterMember> {
- check(::atomix.isInitialized) { "failed to start and join cluster" }
- check(atomix.isRunning) { "cluster is not running" }
-
- return atomix.membershipService.members.filter {
- it.id().id().startsWith(memberPrefix, true)
- }.map { ClusterMember(id = it.id().id(), memberAddress = it.host()) }
- .toSet()
- }
-
- override suspend fun <T> clusterMapStore(name: String): MutableMap<String, T> {
- check(::atomix.isInitialized) { "failed to start and join cluster" }
- return AtomixLibUtils.distributedMapStore<T>(atomix, name)
- }
-
- /** The DistributedLock is a distributed implementation of Java’s Lock.
- * This API provides monotonically increasing, globally unique lock instance identifiers that can be used to
- * determine ordering of multiple concurrent lock holders.
- * DistributedLocks are designed to account for failures within the cluster.
- * When a lock holder crashes or becomes disconnected from the partition by which the lock’s state is controlled,
- * the lock will be released and granted to the next waiting process. *
- */
- override suspend fun clusterLock(name: String): ClusterLock {
- check(::atomix.isInitialized) { "failed to start and join cluster" }
- return ClusterLockImpl(atomix, name)
- }
-
- override suspend fun shutDown(duration: Duration) {
- if (::atomix.isInitialized) {
- val shutDownMilli = duration.toMillis()
- log.info("Received cluster shutdown request, shutdown in ($shutDownMilli)ms")
- delay(shutDownMilli)
- atomix.stop()
- }
- }
-}
-
-open class ClusterLockImpl(private val atomix: Atomix, private val name: String) : ClusterLock {
- val log = logger(ClusterLockImpl::class)
-
- lateinit var distributedLock: DistributedLock
-
- override fun name(): String {
- return distributedLock.name()
- }
-
- override suspend fun lock() {
- distributedLock = AtomixLibUtils.distributedLock(atomix, name)
- distributedLock.lock()
- log.debug("Cluster lock($name) created..")
- }
-
- override suspend fun tryLock(timeout: Long): Boolean {
- distributedLock = AtomixLibUtils.distributedLock(atomix, name)
- return distributedLock.tryLock(Duration.ofMillis(timeout))
- }
-
- override suspend fun unLock() {
- distributedLock.unlock()
- log.debug("Cluster unlock(${name()}) successfully..")
- }
-
- override fun isLocked(): Boolean {
- return distributedLock.isLocked
- }
-
- override fun close() {
- if (::distributedLock.isInitialized) {
- distributedLock.close()
- }
- }
-}
+++ /dev/null
-/*
- * Copyright © 2018-2019 AT&T Intellectual Property.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.onap.ccsdk.cds.blueprintsprocessor.atomix.utils
-
-import com.fasterxml.jackson.databind.JsonNode
-import com.fasterxml.jackson.databind.node.ArrayNode
-import com.fasterxml.jackson.databind.node.MissingNode
-import com.fasterxml.jackson.databind.node.NullNode
-import com.fasterxml.jackson.databind.node.ObjectNode
-import io.atomix.core.Atomix
-import io.atomix.core.lock.AtomicLock
-import io.atomix.core.lock.DistributedLock
-import io.atomix.core.map.DistributedMap
-import io.atomix.protocols.backup.MultiPrimaryProtocol
-import io.atomix.protocols.backup.partition.PrimaryBackupPartitionGroup
-import io.atomix.protocols.raft.partition.RaftPartitionGroup
-import io.atomix.utils.net.Address
-import org.jsoup.nodes.TextNode
-import org.onap.ccsdk.cds.blueprintsprocessor.core.service.ClusterInfo
-import org.onap.ccsdk.cds.controllerblueprints.core.logger
-import org.onap.ccsdk.cds.controllerblueprints.core.normalizedFile
-
-object AtomixLibUtils {
- private val log = logger(AtomixLibUtils::class)
-
- fun configAtomix(filePath: String): Atomix {
- val configFile = normalizedFile(filePath)
- return Atomix.builder(configFile.absolutePath).build()
- }
-
- fun defaultMulticastAtomix(
- clusterInfo: ClusterInfo,
- raftPartitions: Int = 1,
- primaryBackupPartitions: Int = 32
- ): Atomix {
-
- val nodeId = clusterInfo.nodeId
-
- val raftPartitionGroup = RaftPartitionGroup.builder("system")
- .withNumPartitions(raftPartitions)
- .withMembers(clusterInfo.clusterMembers)
- .withDataDirectory(normalizedFile("${clusterInfo.storagePath}/data-$nodeId"))
- .build()
-
- val primaryBackupGroup =
- PrimaryBackupPartitionGroup.builder("data")
- .withNumPartitions(primaryBackupPartitions)
- .build()
-
- return Atomix.builder()
- .withMemberId(nodeId)
- .withAddress(Address.from(clusterInfo.nodeAddress))
- .withManagementGroup(raftPartitionGroup)
- .withPartitionGroups(primaryBackupGroup)
- .withMulticastEnabled()
- .build()
- }
-
- fun <T> distributedMapStore(atomix: Atomix, storeName: String, numBackups: Int = 2): DistributedMap<String, T> {
- check(atomix.isRunning) { "Cluster is not running, couldn't create distributed store($storeName)" }
-
- val protocol = MultiPrimaryProtocol.builder()
- .withBackups(numBackups)
- .build()
-
- return atomix.mapBuilder<String, T>(storeName)
- .withProtocol(protocol)
- .withCacheEnabled()
- .withValueType(JsonNode::class.java)
- .withExtraTypes(
- JsonNode::class.java, TextNode::class.java, ObjectNode::class.java,
- ArrayNode::class.java, NullNode::class.java, MissingNode::class.java
- )
- .build()
- }
-
- fun distributedLock(atomix: Atomix, lockName: String, numBackups: Int = 2): DistributedLock {
- check(atomix.isRunning) { "Cluster is not running, couldn't create distributed lock($lockName)" }
-
- val protocol = MultiPrimaryProtocol.builder()
- .withBackups(numBackups)
- .build()
- return atomix.lockBuilder(lockName)
- .withProtocol(protocol)
- .build()
- }
-
- /** get Atomic distributed lock, to get lock fence information */
- fun atomicLock(atomix: Atomix, lockName: String, numBackups: Int = 2): AtomicLock {
- check(atomix.isRunning) { "Cluster is not running, couldn't create atomic lock($lockName)" }
-
- val protocol = MultiPrimaryProtocol.builder()
- .withBackups(numBackups)
- .build()
-
- return atomix.atomicLockBuilder(lockName)
- .withProtocol(protocol)
- .build()
- }
-}
+++ /dev/null
-/*
- * Copyright © 2018-2019 AT&T Intellectual Property.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.onap.ccsdk.cds.blueprintsprocessor.atomix
-
-import com.fasterxml.jackson.databind.JsonNode
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.async
-import kotlinx.coroutines.awaitAll
-import kotlinx.coroutines.delay
-import kotlinx.coroutines.runBlocking
-import kotlinx.coroutines.withContext
-import org.junit.Before
-import org.junit.Test
-import org.onap.ccsdk.cds.blueprintsprocessor.atomix.service.AtomixBluePrintClusterService
-import org.onap.ccsdk.cds.blueprintsprocessor.core.service.BluePrintClusterService
-import org.onap.ccsdk.cds.blueprintsprocessor.core.service.ClusterInfo
-import org.onap.ccsdk.cds.controllerblueprints.core.asJsonPrimitive
-import org.onap.ccsdk.cds.controllerblueprints.core.deleteNBDir
-import org.onap.ccsdk.cds.controllerblueprints.core.logger
-import kotlin.test.assertNotNull
-import kotlin.test.assertTrue
-
-class AtomixBluePrintClusterServiceTest {
- private val log = logger(AtomixBluePrintClusterServiceTest::class)
-
- @Before
- fun init() {
- runBlocking {
- deleteNBDir("target/cluster")
- }
- }
-
- /** Testing two cluster with distributed map store creation, This is time consuming test case, taks around 10s **/
- @Test
- fun testClusterJoin() {
- runBlocking {
- val bluePrintClusterServiceOne =
- createCluster(arrayListOf(5679, 5680)).toMutableList()
- // val bluePrintClusterServiceTwo = createCluster(arrayListOf(5681, 5682), arrayListOf(5679, 5680))
- // bluePrintClusterServiceOne.addAll(bluePrintClusterServiceTwo)
- val bluePrintClusterService = bluePrintClusterServiceOne[0]
- log.info("Members : ${bluePrintClusterService.allMembers()}")
- log.info("Master(System) Members : ${bluePrintClusterService.masterMember("system")}")
- log.info("Master(Data) Members : ${bluePrintClusterService.masterMember("data")}")
- testDistributedStore(bluePrintClusterServiceOne)
- testDistributedLock(bluePrintClusterServiceOne)
- }
- }
-
- private suspend fun createCluster(
- ports: List<Int>,
- otherClusterPorts: List<Int>? = null
- ): List<BluePrintClusterService> {
-
- return withContext(Dispatchers.Default) {
- val clusterMembers = ports.map { "node-$it" }.toMutableList()
- /** Add the other cluster as members */
- if (!otherClusterPorts.isNullOrEmpty()) {
- val otherClusterMembers = otherClusterPorts.map { "node-$it" }.toMutableList()
- clusterMembers.addAll(otherClusterMembers)
- }
- val deferred = ports.map { port ->
- async(Dispatchers.IO) {
- val nodeId = "node-$port"
- log.info("********** Starting node($nodeId) on port($port)")
- val clusterInfo = ClusterInfo(
- id = "test-cluster", nodeId = nodeId,
- clusterMembers = clusterMembers, nodeAddress = "localhost:$port", storagePath = "target/cluster"
- )
- val atomixClusterService = AtomixBluePrintClusterService()
- atomixClusterService.startCluster(clusterInfo)
- atomixClusterService
- }
- }
- deferred.awaitAll()
- }
- }
-
- private suspend fun testDistributedStore(bluePrintClusterServices: List<BluePrintClusterService>) {
- /** Test Distributed store creation */
- repeat(2) { storeId ->
- val store = bluePrintClusterServices[0].clusterMapStore<JsonNode>(
- "blueprint-runtime-$storeId"
- ).toDistributedMap()
- assertNotNull(store, "failed to get store")
- val store1 = bluePrintClusterServices[1].clusterMapStore<JsonNode>(
- "blueprint-runtime-$storeId"
- ).toDistributedMap()
-
- store1.addListener {
- log.info("Received map event : $it")
- }
- repeat(5) {
- store["key-$storeId-$it"] = "value-$it".asJsonPrimitive()
- }
- delay(10)
- store.close()
- }
- }
-
- private suspend fun testDistributedLock(bluePrintClusterServices: List<BluePrintClusterService>) {
- val lockName = "sample-lock"
- withContext(Dispatchers.IO) {
- val deferred = async {
- executeLock(bluePrintClusterServices[0], "first", lockName)
- }
- val deferred2 = async {
- executeLock(bluePrintClusterServices[0], "second", lockName)
- }
- val deferred3 = async {
- executeLock(bluePrintClusterServices[1], "third", lockName)
- }
- deferred.start()
- deferred2.start()
- deferred3.start()
- }
- }
-
- private suspend fun executeLock(
- bluePrintClusterService: BluePrintClusterService,
- lockId: String,
- lockName: String
- ) {
- log.info("initialising $lockId lock...")
- val distributedLock = bluePrintClusterService.clusterLock(lockName)
- assertNotNull(distributedLock, "failed to create distributed $lockId lock")
- distributedLock.lock()
- assertTrue(distributedLock.isLocked(), "failed to lock $lockId")
- try {
- log.info("locked $lockId process for 5mSec")
- delay(5)
- } finally {
- distributedLock.unLock()
- log.info("$lockId lock released")
- }
- distributedLock.close()
- }
-}
<modules>
<module>processor-core</module>
- <module>atomix-lib</module>
<module>db-lib</module>
<module>rest-lib</module>
<module>dmaap-lib</module>
<description>Blueprints Processor Core</description>
<dependencies>
+ <dependency>
+ <groupId>com.hazelcast</groupId>
+ <artifactId>hazelcast-all</artifactId>
+ </dependency>
<dependency>
<groupId>org.onap.ccsdk.cds.blueprintsprocessor</groupId>
<artifactId>blueprint-proto</artifactId>
* limitations under the License.
*/
-package org.onap.ccsdk.cds.blueprintsprocessor.atomix
+package org.onap.ccsdk.cds.blueprintsprocessor.core.cluster
-import org.onap.ccsdk.cds.blueprintsprocessor.atomix.service.AtomixBluePrintClusterService
+import com.hazelcast.cluster.Member
import org.onap.ccsdk.cds.blueprintsprocessor.core.service.BluePrintClusterService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.service.ClusterMember
import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants
import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService
-import org.springframework.context.annotation.Configuration
-
-@Configuration
-open class BluePrintAtomixLibConfiguration
/**
- * Exposed Dependency Service by this Atomix Lib Module
+ * Exposed Dependency Service by this Hazlecast Lib Module
*/
fun BluePrintDependencyService.clusterService(): BluePrintClusterService =
- instance(AtomixBluePrintClusterService::class)
+ instance(HazlecastClusterService::class)
/** Optional Cluster Service, returns only if Cluster is enabled */
fun BluePrintDependencyService.optionalClusterService(): BluePrintClusterService? {
BluePrintDependencyService.clusterService()
} else null
}
+
+/** Extension to convert Hazelcast Member to Blueprints Cluster Member */
+fun Member.toClusterMember(): ClusterMember {
+ val memberName: String = this.getAttribute(BluePrintConstants.PROPERTY_CLUSTER_NODE_ID) ?: this.uuid.toString()
+ return ClusterMember(
+ id = this.uuid.toString(),
+ name = memberName,
+ memberAddress = this.address.toString()
+ )
+}
--- /dev/null
+/*
+ * Copyright © 2018-2019 AT&T Intellectual Property.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onap.ccsdk.cds.blueprintsprocessor.core.cluster
+
+import com.hazelcast.client.HazelcastClient
+import com.hazelcast.client.config.ClientConfig
+import com.hazelcast.client.config.YamlClientConfigBuilder
+import com.hazelcast.cluster.Member
+import com.hazelcast.cluster.MembershipEvent
+import com.hazelcast.cluster.MembershipListener
+import com.hazelcast.config.Config
+import com.hazelcast.config.FileSystemYamlConfig
+import com.hazelcast.config.MemberAttributeConfig
+import com.hazelcast.core.Hazelcast
+import com.hazelcast.core.HazelcastInstance
+import com.hazelcast.cp.lock.FencedLock
+import com.hazelcast.scheduledexecutor.IScheduledExecutorService
+import kotlinx.coroutines.delay
+import org.onap.ccsdk.cds.blueprintsprocessor.core.service.BluePrintClusterService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.service.ClusterInfo
+import org.onap.ccsdk.cds.blueprintsprocessor.core.service.ClusterLock
+import org.onap.ccsdk.cds.blueprintsprocessor.core.service.ClusterMember
+import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants
+import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+import org.onap.ccsdk.cds.controllerblueprints.core.normalizedFile
+import org.onap.ccsdk.cds.controllerblueprints.core.utils.ClusterUtils
+import org.springframework.stereotype.Service
+import java.time.Duration
+import java.util.concurrent.TimeUnit
+
+@Service
+open class HazlecastClusterService : BluePrintClusterService {
+
+ private val log = logger(HazlecastClusterService::class)
+ lateinit var hazelcast: HazelcastInstance
+ var joinedClient = false
+ var joinedLite = false
+
+ override suspend fun <T> startCluster(configuration: T) {
+ /** Get the Hazelcast Cliet or Server instance */
+ hazelcast =
+ when (configuration) {
+ is Config -> {
+ joinedLite = configuration.isLiteMember
+ Hazelcast.newHazelcastInstance(configuration)
+ }
+ is ClientConfig -> {
+ joinedClient = true
+ HazelcastClient.newHazelcastClient(configuration)
+ }
+ is ClusterInfo -> {
+
+ System.setProperty(BluePrintConstants.PROPERTY_CLUSTER_ID, configuration.id)
+ System.setProperty(BluePrintConstants.PROPERTY_CLUSTER_NODE_ID, configuration.nodeId)
+
+ val memberAttributeConfig = MemberAttributeConfig()
+ memberAttributeConfig.setAttribute(
+ BluePrintConstants.PROPERTY_CLUSTER_NODE_ID,
+ configuration.nodeId
+ )
+
+ val configFile = configuration.configFile
+ /** Check file exists */
+ val clusterConfigFile = normalizedFile(configuration.configFile)
+ check(clusterConfigFile.absolutePath.endsWith("yaml", true)) {
+ "couldn't understand cluster config file(${configuration.configFile}) format, it should be yaml"
+ }
+ check(clusterConfigFile.exists()) {
+ "couldn't file cluster configuration file(${clusterConfigFile.absolutePath})"
+ }
+ log.info("****** Cluster configuration file(${clusterConfigFile.absolutePath}) ****")
+
+ /** Hazelcast Client from config file */
+ if (configuration.joinAsClient) {
+ /** Set the configuration file to system properties, so that Hazelcast will read automatically */
+ System.setProperty("hazelcast.client.config", clusterConfigFile.absolutePath)
+ joinedClient = true
+ val hazelcastClientConfiguration = YamlClientConfigBuilder().build()
+ hazelcastClientConfiguration.properties = configuration.properties
+ HazelcastClient.newHazelcastClient(hazelcastClientConfiguration)
+ } else {
+ /** Hazelcast Server from config file */
+ val hazelcastServerConfiguration = FileSystemYamlConfig(normalizedFile(configFile))
+ hazelcastServerConfiguration.properties = configuration.properties
+ hazelcastServerConfiguration.memberAttributeConfig = memberAttributeConfig
+ joinedLite = hazelcastServerConfiguration.isLiteMember
+ Hazelcast.newHazelcastInstance(hazelcastServerConfiguration)
+ }
+ }
+ else -> {
+ throw BluePrintProcessorException("couldn't understand the cluster configuration")
+ }
+ }
+
+ /** Add the Membership Listeners */
+ hazelcast.cluster.addMembershipListener(BlueprintsClusterMembershipListener(this))
+ log.info(
+ "Cluster(${hazelcast.config.clusterName}) node(${hazelcast.name}) created successfully...."
+ )
+ }
+
+ override fun isClient(): Boolean {
+ return joinedClient
+ }
+
+ override fun isLiteMember(): Boolean {
+ return joinedLite
+ }
+
+ override fun clusterJoined(): Boolean {
+ return hazelcast.lifecycleService.isRunning
+ }
+
+ override suspend fun masterMember(partitionGroup: String): ClusterMember {
+ check(::hazelcast.isInitialized) { "failed to start and join cluster" }
+ return hazelcast.cluster.members.first().toClusterMember()
+ }
+
+ override suspend fun allMembers(): Set<ClusterMember> {
+ check(::hazelcast.isInitialized) { "failed to start and join cluster" }
+ return hazelcast.cluster.members.map { it.toClusterMember() }.toSet()
+ }
+
+ override suspend fun applicationMembers(appName: String): Set<ClusterMember> {
+ check(::hazelcast.isInitialized) { "failed to start and join cluster" }
+ return hazelcastApplicationMembers(appName).mapNotNull { it.value.toClusterMember() }.toSet()
+ }
+
+ override suspend fun <T> clusterMapStore(name: String): MutableMap<String, T> {
+ check(::hazelcast.isInitialized) { "failed to start and join cluster" }
+ return hazelcast.getMap<String, T>(name)
+ }
+
+ /**
+ * The DistributedLock is a distributed implementation of Java’s Lock.
+ * This API provides monotonically increasing, globally unique lock instance identifiers that can be used to
+ * determine ordering of multiple concurrent lock holders.
+ * DistributedLocks are designed to account for failures within the cluster.
+ * When a lock holder crashes or becomes disconnected from the partition by which the lock’s state is controlled,
+ * the lock will be released and granted to the next waiting process.
+ */
+ override suspend fun clusterLock(name: String): ClusterLock {
+ check(::hazelcast.isInitialized) { "failed to start and join cluster" }
+ return ClusterLockImpl(hazelcast, name)
+ }
+
+ /** Return interface may change and it will be included in BluePrintClusterService */
+ @UseExperimental
+ suspend fun clusterScheduler(name: String): IScheduledExecutorService {
+ check(::hazelcast.isInitialized) { "failed to start and join cluster" }
+ return hazelcast.getScheduledExecutorService(name)
+ }
+
+ override suspend fun shutDown(duration: Duration) {
+ if (::hazelcast.isInitialized && clusterJoined()) {
+ delay(duration.toMillis())
+ hazelcast.lifecycleService.terminate()
+ }
+ }
+
+ /** Utils */
+ suspend fun myHazelcastApplicationMembers(): Map<String, Member> {
+ check(::hazelcast.isInitialized) { "failed to start and join cluster" }
+ check(!isClient()) { "not supported for cluster client members." }
+ return hazelcastApplicationMembers(ClusterUtils.applicationName())
+ }
+
+ suspend fun hazelcastApplicationMembers(appName: String): Map<String, Member> {
+ check(::hazelcast.isInitialized) { "failed to start and join cluster" }
+ val applicationMembers: MutableMap<String, Member> = hashMapOf()
+ hazelcast.cluster.members.map { member ->
+ val memberName: String = member.getAttribute(BluePrintConstants.PROPERTY_CLUSTER_NODE_ID)
+ if (memberName.startsWith(appName, true)) {
+ applicationMembers[memberName] = member
+ }
+ }
+ return applicationMembers
+ }
+}
+
+open class BlueprintsClusterMembershipListener(val hazlecastClusterService: HazlecastClusterService) :
+ MembershipListener {
+ private val log = logger(BlueprintsClusterMembershipListener::class)
+
+ override fun memberRemoved(membershipEvent: MembershipEvent) {
+ log.info("${hazlecastClusterService.hazelcast.cluster.localMember} : Member Removed: $membershipEvent")
+ }
+
+ override fun memberAdded(membershipEvent: MembershipEvent) {
+ log.info("${hazlecastClusterService.hazelcast.cluster.localMember} : Member Added : $membershipEvent")
+ }
+}
+
+open class ClusterLockImpl(private val hazelcast: HazelcastInstance, private val name: String) : ClusterLock {
+ private val log = logger(ClusterLockImpl::class)
+
+ lateinit var distributedLock: FencedLock
+
+ override fun name(): String {
+ return distributedLock.name
+ }
+
+ override suspend fun lock() {
+ distributedLock = hazelcast.cpSubsystem.getLock(name)
+ distributedLock.lock()
+ log.trace("Cluster lock($name) created..")
+ }
+
+ override suspend fun tryLock(timeout: Long): Boolean {
+ distributedLock = hazelcast.cpSubsystem.getLock(name)
+ return distributedLock.tryLock(timeout, TimeUnit.MILLISECONDS)
+ }
+
+ override suspend fun unLock() {
+ distributedLock.unlock()
+ log.trace("Cluster unlock(${name()}) successfully..")
+ }
+
+ override fun isLocked(): Boolean {
+ return distributedLock.isLocked
+ }
+
+ override suspend fun fenceLock(): String {
+ distributedLock = hazelcast.cpSubsystem.getLock(name)
+ val fence = distributedLock.lockAndGetFence()
+ log.trace("Cluster lock($name) fence($fence) created..")
+ return fence.toString()
+ }
+
+ override suspend fun tryFenceLock(timeout: Long): String {
+ distributedLock = hazelcast.cpSubsystem.getLock(name)
+ return distributedLock.tryLockAndGetFence(timeout, TimeUnit.MILLISECONDS).toString()
+ }
+
+ override fun close() {
+ }
+}
package org.onap.ccsdk.cds.blueprintsprocessor.core.service
import java.time.Duration
+import java.util.Properties
interface BluePrintClusterService {
/** Start the cluster with [clusterInfo], By default clustering service is disabled.
* Application module has to start cluster */
- suspend fun startCluster(clusterInfo: ClusterInfo)
+ suspend fun <T> startCluster(configuration: T)
fun clusterJoined(): Boolean
+ fun isClient(): Boolean
+
+ fun isLiteMember(): Boolean
+
/** Returns [partitionGroup] master member */
suspend fun masterMember(partitionGroup: String): ClusterMember
/** Returns all the data cluster members */
suspend fun allMembers(): Set<ClusterMember>
- /** Returns data cluster members starting with prefix */
- suspend fun clusterMembersForPrefix(memberPrefix: String): Set<ClusterMember>
+ /**
+ * Returns application cluster members for [appName] joined as server or lite member,
+ * Node joined as client won't be visible. Here the assumption is node-id is combination of
+ * application id and replica number, for an example Application cds-cluster then the node ids will be
+ * cds-cluster-1, cds-cluster-2, cds-cluster-3
+ */
+ suspend fun applicationMembers(appName: String): Set<ClusterMember>
/** Create and get or get the distributed data map store with [name] */
suspend fun <T> clusterMapStore(name: String): MutableMap<String, T>
data class ClusterInfo(
val id: String,
- var configFile: String? = null,
val nodeId: String,
- val nodeAddress: String,
- var clusterMembers: List<String>,
- var storagePath: String
+ var joinAsClient: Boolean = false,
+ var properties: Properties?,
+ var configFile: String
)
-data class ClusterMember(val id: String, val memberAddress: String?, val state: String? = null)
+data class ClusterMember(
+ val id: String,
+ val name: String,
+ val memberAddress: String?,
+ val state: String? = null
+)
interface ClusterLock {
fun name(): String
suspend fun lock()
+ suspend fun fenceLock(): String
suspend fun tryLock(timeout: Long): Boolean
+ suspend fun tryFenceLock(timeout: Long): String
suspend fun unLock()
fun isLocked(): Boolean
fun close()
--- /dev/null
+/*
+ * Copyright © 2018-2019 AT&T Intellectual Property.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onap.ccsdk.cds.blueprintsprocessor.core.cluster
+
+import com.fasterxml.jackson.databind.JsonNode
+import com.hazelcast.client.config.YamlClientConfigBuilder
+import com.hazelcast.cluster.Member
+import com.hazelcast.config.FileSystemYamlConfig
+import com.hazelcast.map.IMap
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withContext
+import org.junit.Test
+import org.onap.ccsdk.cds.blueprintsprocessor.core.service.BluePrintClusterService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.service.ClusterInfo
+import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants
+import org.onap.ccsdk.cds.controllerblueprints.core.asJsonPrimitive
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+import org.onap.ccsdk.cds.controllerblueprints.core.normalizedFile
+import java.io.Serializable
+import java.time.Duration
+import java.util.Properties
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+import kotlin.test.assertTrue
+
+class HazlecastClusterServiceTest {
+ private val log = logger(HazlecastClusterServiceTest::class)
+ private val clusterSize = 3
+
+ @Test
+ fun testClientFileSystemYamlConfig() {
+ System.setProperty(BluePrintConstants.PROPERTY_CLUSTER_ID, "test-cluster")
+ System.setProperty(BluePrintConstants.PROPERTY_CLUSTER_NODE_ID, "node-1234")
+ System.setProperty(
+ "hazelcast.client.config",
+ normalizedFile("./src/test/resources/hazelcast/hazelcast-client.yaml").absolutePath
+ )
+ val config = YamlClientConfigBuilder().build()
+ assertNotNull(config)
+ assertEquals("test-cluster", config.clusterName)
+ assertEquals("node-1234", config.instanceName)
+ }
+
+ @Test
+ fun testServerFileSystemYamlConfig() {
+ System.setProperty(BluePrintConstants.PROPERTY_CLUSTER_ID, "test-cluster")
+ System.setProperty(BluePrintConstants.PROPERTY_CLUSTER_NODE_ID, "node-1234")
+ val configFile = normalizedFile("./src/test/resources/hazelcast/hazelcast.yaml")
+ val config = FileSystemYamlConfig(configFile)
+ assertNotNull(config)
+ assertEquals("test-cluster", config.clusterName)
+ assertEquals("node-1234", config.instanceName)
+ }
+
+ @Test
+ fun testClusterJoin() {
+ runBlocking {
+ val bluePrintClusterServiceOne =
+ createCluster(arrayListOf(5679, 5680, 5681)).toMutableList()
+ // delay(1000)
+ // Join as Hazlecast Management Node
+ // val bluePrintClusterServiceTwo = createCluster(arrayListOf(5682), true)
+ // val bluePrintClusterServiceTwo = createCluster(arrayListOf(5682), false)
+ // bluePrintClusterServiceOne.addAll(bluePrintClusterServiceTwo)
+ printReachableMembers(bluePrintClusterServiceOne)
+ testDistributedStore(bluePrintClusterServiceOne)
+ testDistributedLock(bluePrintClusterServiceOne)
+
+ // executeScheduler(bluePrintClusterServiceOne[0])
+ // delay(1000)
+ // Shutdown
+ shutdown(bluePrintClusterServiceOne)
+ }
+ }
+
+ private suspend fun createCluster(
+ ports: List<Int>,
+ joinAsClient: Boolean? = false
+ ): List<BluePrintClusterService> {
+
+ return withContext(Dispatchers.Default) {
+ val deferred = ports.map { port ->
+ async(Dispatchers.IO) {
+ val nodeId = "node-$port"
+ log.info("********** Starting node($nodeId) on port($port)")
+ val properties = Properties()
+ properties["hazelcast.logging.type"] = "slf4j"
+ val clusterInfo =
+ if (joinAsClient!!) {
+ ClusterInfo(
+ id = "test-cluster", nodeId = nodeId, joinAsClient = true,
+ configFile = "./src/test/resources/hazelcast/hazelcast-client.yaml",
+ properties = properties
+ )
+ } else {
+ ClusterInfo(
+ id = "test-cluster", nodeId = nodeId, joinAsClient = false,
+ configFile = "./src/test/resources/hazelcast/hazelcast-$port.yaml",
+ properties = properties
+ )
+ }
+ val hazlecastClusterService = HazlecastClusterService()
+ hazlecastClusterService.startCluster(clusterInfo)
+ hazlecastClusterService
+ }
+ }
+ deferred.awaitAll()
+ }
+ }
+
+ private suspend fun shutdown(bluePrintClusterServices: List<BluePrintClusterService>) {
+ bluePrintClusterServices.forEach { bluePrintClusterService ->
+ bluePrintClusterService.shutDown(Duration.ofMillis(10))
+ }
+ }
+
+ private suspend fun testDistributedStore(bluePrintClusterServices: List<BluePrintClusterService>) {
+ /** Test Distributed store creation */
+ repeat(2) { storeId ->
+ val store = bluePrintClusterServices[0].clusterMapStore<JsonNode>(
+ "blueprint-runtime-$storeId"
+ ) as IMap
+ assertNotNull(store, "failed to get store")
+ repeat(5) {
+ store["key-$storeId-$it"] = "value-$it".asJsonPrimitive()
+ }
+
+ val store1 = bluePrintClusterServices[1].clusterMapStore<JsonNode>(
+ "blueprint-runtime-$storeId"
+ ) as IMap
+
+ store1.values.map {
+ log.trace("Received map event : $it")
+ }
+ delay(5)
+ store.clear()
+ }
+ }
+
+ private suspend fun testDistributedLock(bluePrintClusterServices: List<BluePrintClusterService>) {
+ val lockName = "sample-lock"
+ withContext(Dispatchers.IO) {
+ val deferred = async {
+ executeLock(bluePrintClusterServices[0], "first", lockName)
+ }
+ val deferred2 = async {
+ executeLock(bluePrintClusterServices[0], "second", lockName)
+ }
+ val deferred3 = async {
+ executeLock(bluePrintClusterServices[2], "third", lockName)
+ }
+ deferred.start()
+ deferred2.start()
+ deferred3.start()
+ }
+ }
+
+ private suspend fun executeLock(
+ bluePrintClusterService: BluePrintClusterService,
+ lockId: String,
+ lockName: String
+ ) {
+ log.info("initialising $lockId lock...")
+ val distributedLock = bluePrintClusterService.clusterLock(lockName)
+ assertNotNull(distributedLock, "failed to create distributed $lockId lock")
+ distributedLock.lock()
+ assertTrue(distributedLock.isLocked(), "failed to lock $lockId")
+ try {
+ log.info("locked $lockId process for 5mSec")
+ delay(5)
+ } finally {
+ distributedLock.unLock()
+ log.info("$lockId lock released")
+ }
+ distributedLock.close()
+ }
+
+ private suspend fun executeScheduler(bluePrintClusterService: BluePrintClusterService) {
+ log.info("initialising ...")
+ val hazlecastClusterService = bluePrintClusterService as HazlecastClusterService
+
+ val memberNameMap = bluePrintClusterService.clusterMapStore<Member>("member-name-map") as IMap
+ assertEquals(3, memberNameMap.size, "failed to match member size")
+ memberNameMap.forEach { (key, value) -> log.info("nodeId($key), Member($value)") }
+ val scheduler = hazlecastClusterService.clusterScheduler("cleanup")
+ // scheduler.scheduleOnAllMembers(SampleSchedulerTask(), 0, TimeUnit.SECONDS)
+ // scheduler.scheduleOnKeyOwnerAtFixedRate(SampleSchedulerTask(), "node-5680",0, 1, TimeUnit.SECONDS)
+ // scheduler.scheduleAtFixedRate(SampleSchedulerTask(), 0, 1, TimeUnit.SECONDS)
+ // scheduler.scheduleOnAllMembersAtFixedRate(SampleSchedulerTask(), 0, 5, TimeUnit.SECONDS)
+ }
+
+ private suspend fun printReachableMembers(bluePrintClusterServices: List<BluePrintClusterService>) {
+ bluePrintClusterServices.forEach { bluePrintClusterService ->
+ val hazlecastClusterService = bluePrintClusterService as HazlecastClusterService
+ val hazelcast = hazlecastClusterService.hazelcast
+ val self = if (!bluePrintClusterService.isClient()) hazelcast.cluster.localMember else null
+ val master = hazlecastClusterService.masterMember("system").memberAddress
+ val members = hazlecastClusterService.allMembers().map { it.memberAddress }
+ log.info("Cluster Members for($self): master($master) Members($members)")
+ }
+
+ val applicationMembers = bluePrintClusterServices[0].applicationMembers("node-56")
+ assertEquals(clusterSize, applicationMembers.size, "failed to match applications member size")
+ log.info("Cluster applicationMembers ($applicationMembers)")
+ }
+}
+
+open class SampleSchedulerTask : Runnable, Serializable {
+ private val log = logger(SampleSchedulerTask::class)
+ override fun run() {
+ log.info("I am scheduler action")
+ }
+}
--- /dev/null
+hazelcast:
+ cluster-name: ${CLUSTER_ID}
+ instance-name: node-5679
+ lite-member:
+ enabled: false
+ cp-subsystem:
+ cp-member-count: 3
+ group-size: 3
+# network:
+# join:
+# multicast:
+# enabled: false
+# kubernetes:
+# enabled: true
+# namespace: MY-KUBERNETES-NAMESPACE
+# service-name: MY-SERVICE-NAME
+# service-label-name: MY-SERVICE-LABEL-NAME
+# service-label-value: MY-SERVICE-LABEL-VALUE
\ No newline at end of file
--- /dev/null
+hazelcast:
+ cluster-name: ${CLUSTER_ID}
+ instance-name: node-5680
+ lite-member:
+ enabled: false
+ cp-subsystem:
+ cp-member-count: 3
+ group-size: 3
+# network:
+# join:
+# multicast:
+# enabled: false
+# kubernetes:
+# enabled: true
+# namespace: MY-KUBERNETES-NAMESPACE
+# service-name: MY-SERVICE-NAME
+# service-label-name: MY-SERVICE-LABEL-NAME
+# service-label-value: MY-SERVICE-LABEL-VALUE
\ No newline at end of file
--- /dev/null
+hazelcast:
+ cluster-name: ${CLUSTER_ID}
+ instance-name: node-5681
+ lite-member:
+ enabled: false
+ cp-subsystem:
+ cp-member-count: 3
+ group-size: 3
+# network:
+# join:
+# multicast:
+# enabled: false
+# kubernetes:
+# enabled: true
+# namespace: MY-KUBERNETES-NAMESPACE
+# service-name: MY-SERVICE-NAME
+# service-label-name: MY-SERVICE-LABEL-NAME
+# service-label-value: MY-SERVICE-LABEL-VALUE
\ No newline at end of file
--- /dev/null
+hazelcast:
+ cluster-name: ${CLUSTER_ID}
+ instance-name: node-5682
+ lite-member:
+ enabled: true
+ cp-subsystem:
+ cp-member-count: 3
+ group-size: 3
+# network:
+# join:
+# multicast:
+# enabled: false
+# kubernetes:
+# enabled: true
+# namespace: MY-KUBERNETES-NAMESPACE
+# service-name: MY-SERVICE-NAME
+# service-label-name: MY-SERVICE-LABEL-NAME
+# service-label-value: MY-SERVICE-LABEL-VALUE
\ No newline at end of file
--- /dev/null
+hazelcast-client:
+ cluster-name: ${CLUSTER_ID}
+ instance-name: ${CLUSTER_NODE_ID}
+
+ network:
+ cluster-members:
+ - 127.0.0.1:5701
+# kubernetes:
+# enabled: true
+# namespace: MY-KUBERNETES-NAMESPACE
+# service-name: MY-SERVICE-NAME
+# service-label-name: MY-SERVICE-LABEL-NAME
+# service-label-value: MY-SERVICE-LABEL-VALUE
--- /dev/null
+hazelcast:
+ cluster-name: ${CLUSTER_ID}
+ instance-name: ${CLUSTER_NODE_ID}
+ lite-member:
+ enabled: true
+ cp-subsystem:
+ cp-member-count: 3
+ group-size: 3
+# network:
+# join:
+# multicast:
+# enabled: false
+# kubernetes:
+# enabled: true
+# namespace: MY-KUBERNETES-NAMESPACE
+# service-name: MY-SERVICE-NAME
+# service-label-name: MY-SERVICE-LABEL-NAME
+# service-label-value: MY-SERVICE-LABEL-VALUE
\ No newline at end of file
<!--
- ~ Copyright © 2017-2018 AT&T Intellectual Property.
+ ~ Copyright © 2018-2019 AT&T Intellectual Property.
~
~ Licensed under the Apache License, Version 2.0 (the "License");
~ you may not use this file except in compliance with the License.
<logger name="org.springframework.test" level="warn"/>
<logger name="org.springframework" level="warn"/>
<logger name="org.hibernate" level="info"/>
- <logger name="io.atomix" level="warn"/>
+ <logger name="com.hazelcast" level="warn"/>
<logger name="org.onap.ccsdk.cds.blueprintsprocessor" level="info"/>
<root level="warn">
<properties>
<sli.version>${ccsdk.sli.core.version}</sli.version>
- <!-- Override CDS version from parent to be project.version -->
- <ccsdk.cds.version>${project.version}</ccsdk.cds.version>
+ <!-- Override CDS version from parent to be project.version -->
+ <ccsdk.cds.version>${project.version}</ccsdk.cds.version>
<dmaap.client.version>1.1.5</dmaap.client.version>
<!-- Should be using released artifact as soon as available: -->
<!-- https://github.com/springfox/springfox/milestone/44 -->
<springfox.swagger2.version>2.9.2</springfox.swagger2.version>
<eelf.version>1.0.0</eelf.version>
<onap.logger.slf4j>1.2.2</onap.logger.slf4j>
+ <hazelcast.version>4.0</hazelcast.version>
<h2database.version>1.4.197</h2database.version>
<powermock.version>1.7.4</powermock.version>
<version>${nats.streaming.version}</version>
</dependency>
- <!-- Atomix -->
+ <!-- Hazelcast -->
<dependency>
- <groupId>io.atomix</groupId>
- <artifactId>atomix</artifactId>
- <version>${atomix.version}</version>
- </dependency>
- <dependency>
- <groupId>io.atomix</groupId>
- <artifactId>atomix-raft</artifactId>
- <version>${atomix.version}</version>
- </dependency>
- <dependency>
- <groupId>io.atomix</groupId>
- <artifactId>atomix-primary-backup</artifactId>
- <version>${atomix.version}</version>
- </dependency>
- <dependency>
- <groupId>io.atomix</groupId>
- <artifactId>atomix-gossip</artifactId>
- <version>${atomix.version}</version>
+ <groupId>com.hazelcast</groupId>
+ <artifactId>hazelcast-all</artifactId>
+ <version>${hazelcast.version}</version>
</dependency>
<!-- Adaptors -->
<artifactId>processor-core</artifactId>
<version>${ccsdk.cds.version}</version>
</dependency>
- <dependency>
- <groupId>org.onap.ccsdk.cds.blueprintsprocessor</groupId>
- <artifactId>atomix-lib</artifactId>
- <version>${ccsdk.cds.version}</version>
- </dependency>
<dependency>
<groupId>org.onap.ccsdk.cds.blueprintsprocessor</groupId>
<artifactId>db-lib</artifactId>
<modules>
<module>ms</module>
- <module>cds-ui</module>
+ <!-- <module>cds-ui</module>-->
<module>components</module>
</modules>
<!--Only include java and kt files to the scan-->
<sonar.inclusions>**/*.java,**/*.kt</sonar.inclusions>
<!--Specify path to load jacoco XLM report, as Sonar can't load Kotlin coverage from binary report-->
- <sonar.coverage.jacoco.xmlReportPaths>${project.reporting.outputDirectory}/jacoco-ut/jacoco.xml</sonar.coverage.jacoco.xmlReportPaths>
+ <sonar.coverage.jacoco.xmlReportPaths>${project.reporting.outputDirectory}/jacoco-ut/jacoco.xml
+ </sonar.coverage.jacoco.xmlReportPaths>
<jacoco.version>0.8.3</jacoco.version>
<!-- Properties for POM Format -->
Code Review / ccsdk / cds.git / commitdiff