홍길동 / onos

Go to a project
Toggle navigation Toggle navigation pinning
Madan Jampani
Committed by Gerrit Code Review
ab7e7cd2 1 parent d7558938

First round of ClusterMetadata improvements: 

    Introduced a PartitionId type for identifying partitions
    Introduced a admin service for making metadata updates
    Update cluster.json format to specify all partitions (including p0) and changed partitionId to be an int.

Change-Id: Ia0617f1ed0ce886680dcee4f5396a4bbdfa225da
Inline Side-by-side
Showing 14 changed files with 380 additions and 147 deletions
......@@ -20,6 +20,8 @@ import java.util.Collection;
import java.util.Set;
import java.util.stream.Collectors;
import org.apache.commons.collections.CollectionUtils;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Verify.verifyNotNull;
import static com.google.common.base.Verify.verify;
......@@ -32,7 +34,7 @@ import com.google.common.collect.Sets;
/**
* Cluster metadata.
* <p>
* Metadata specifies the attributes that define a ONOS cluster and comprises the collection
* Metadata specifies how a ONOS cluster is constituted and is made up of the collection
* of {@link org.onosproject.cluster.ControllerNode nodes} and the collection of data
* {@link org.onosproject.cluster.Partition partitions}.
*/
......@@ -71,7 +73,8 @@ public final class ClusterMetadata {
}
/**
* Returns the collection of data {@link org.onosproject.cluster.Partition partitions} that make up the cluster.
* Returns the collection of {@link org.onosproject.cluster.Partition partitions} that make
* up the cluster.
* @return collection of partitions.
*/
public Collection<Partition> getPartitions() {
......@@ -93,7 +96,7 @@ public final class ClusterMetadata {
}
/*
* Provide a deep quality check of the meta data (non-Javadoc)
* Provide a deep equality check of the cluster metadata (non-Javadoc)
*
* @see java.lang.Object#equals(java.lang.Object)
*/
......@@ -146,7 +149,7 @@ public final class ClusterMetadata {
}
/**
* Sets the collection of data partitions, returning the cluster metadata builder for method chaining.
* Sets the partitions, returning the cluster metadata builder for method chaining.
* @param partitions collection of partitions
* @return this cluster metadata builder
*/
......@@ -171,10 +174,8 @@ public final class ClusterMetadata {
*/
private void verifyMetadata() {
verifyNotNull(metadata.getName(), "Cluster name must be specified");
verifyNotNull(metadata.getNodes(), "Cluster nodes must be specified");
verifyNotNull(metadata.getPartitions(), "Cluster partitions must be specified");
verify(!metadata.getNodes().isEmpty(), "Cluster nodes must not be empty");
verify(!metadata.getPartitions().isEmpty(), "Cluster nodes must not be empty");
verify(CollectionUtils.isEmpty(metadata.getNodes()), "Cluster nodes must be specified");
verify(CollectionUtils.isEmpty(metadata.getPartitions()), "Cluster partitions must be specified");
// verify that partitions are constituted from valid cluster nodes.
boolean validPartitions = Collections2.transform(metadata.getNodes(), ControllerNode::id)
......@@ -185,4 +186,4 @@ public final class ClusterMetadata {
verify(validPartitions, "Partition locations must be valid cluster nodes");
}
}
}
}
\ No newline at end of file
......
/*
* Copyright 2016 Open Networking Laboratory
*
* 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.onosproject.cluster;
/**
* Service for making updates to {@link ClusterMetadata cluster metadata}.
*/
public interface ClusterMetadataAdminService {
/**
* Updates the cluster metadata.
* @param metadata new metadata
*/
void setClusterMetadata(ClusterMetadata metadata);
}
......@@ -15,10 +15,13 @@
*/
package org.onosproject.cluster;
import org.onosproject.event.ListenerService;
/**
* Service for obtaining metadata information about the cluster.
* Service for accessing {@link ClusterMetadata cluster metadata}.
*/
public interface ClusterMetadataService {
public interface ClusterMetadataService
extends ListenerService<ClusterMetadataEvent, ClusterMetadataEventListener> {
/**
* Returns the current cluster metadata.
......@@ -27,13 +30,7 @@ public interface ClusterMetadataService {
ClusterMetadata getClusterMetadata();
/**
* Updates the cluster metadata.
* @param metadata new metadata
*/
void setClusterMetadata(ClusterMetadata metadata);
/**
* Returns the local controller node representing this instance.
* Returns the {@link ControllerNode controller node} representing this instance.
* @return local controller node
*/
ControllerNode getLocalNode();
......
......@@ -21,14 +21,14 @@ import org.onosproject.store.Store;
import org.onosproject.store.service.Versioned;
/**
* Manages persistence of cluster metadata; not intended for direct use.
* Manages persistence of {@link ClusterMetadata cluster metadata}; not intended for direct use.
*/
public interface ClusterMetadataStore extends Store<ClusterMetadataEvent, ClusterMetadataStoreDelegate> {
/**
* Returns the cluster metadata.
* <p>
* The returned metadata is versioned to aid determining if a metadata instance is more recent than another.
* The retuned metadata is encapsulated as a {@link Versioned versioned} and therefore has a specific version.
* @return cluster metadata
*/
Versioned<ClusterMetadata> getClusterMetadata();
......@@ -39,39 +39,38 @@ public interface ClusterMetadataStore extends Store<ClusterMetadataEvent, Cluste
*/
void setClusterMetadata(ClusterMetadata metadata);
// TODO: The below methods should move to a separate store interface that is responsible for
// tracking cluster partition operational state.
/**
* Sets a controller node as an active member of a partition.
* Adds a controller node to the list of active members for a partition.
* <p>
* Active members are those replicas that are up to speed with the rest of the system and are
* usually capable of participating in the replica state management activities in accordance with
* the data consistency and replication protocol in use.
* Active members of a partition are those that are actively participating
* in the data replication protocol being employed. When a node first added
* to a partition, it is in a passive or catch up mode where it attempts to
* bring it self up to speed with other active members in the partition.
* @param partitionId partition identifier
* @param nodeId id of controller node
* @param nodeId identifier of controller node
*/
void setActiveReplica(String partitionId, NodeId nodeId);
void addActivePartitionMember(PartitionId partitionId, NodeId nodeId);
/**
* Removes a controller node as an active member for a partition.
* <p>
* Active members are those replicas that are up to speed with the rest of the system and are
* usually capable of participating in the replica state management activities in accordance with
* the data consistency and replication protocol in use.
* Removes a controller node from the list of active members for a partition.
* @param partitionId partition identifier
* @param nodeId id of controller node
*/
void unsetActiveReplica(String partitionId, NodeId nodeId);
void removeActivePartitionMember(PartitionId partitionId, NodeId nodeId);
/**
* Returns the collection of controller nodes that are the active replicas for a partition.
* Returns the collection of controller nodes that are the active members for a partition.
* <p>
* Active members of a partition are typically those that are actively
* participating in the data replication protocol being employed. When
* a node first added to a partition, it is in a passive or catch up mode where
* it attempts to bring it self up to speed with other active members in the partition.
* <p>
* Active members are those replicas that are up to speed with the rest of the system and are
* usually capable of participating in the replica state management activities in accordance with
* the data consistency and replication protocol in use.
* <b>Note:</b>If is possible for this list to different from the list of partition members
* specified by cluster meta data. The discrepancy can arise due to the fact that
* adding/removing members from a partition requires a data hand-off mechanism to complete.
* @param partitionId partition identifier
* @return identifiers of controller nodes that are the active replicas
* @return identifiers of controller nodes that are active members
*/
Collection<NodeId> getActiveReplicas(String partitionId);
Collection<NodeId> getActivePartitionMembers(PartitionId partitionId);
}
\ No newline at end of file
......
/*
* Copyright 2015 Open Networking Laboratory
* Copyright 2016 Open Networking Laboratory
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
......@@ -15,77 +15,23 @@
*/
package org.onosproject.cluster;
import java.util.Arrays;
import java.util.Collection;
import java.util.Set;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Sets;
import static com.google.common.base.Preconditions.checkNotNull;
/**
* A data partition.
* <p>
* Partition represents a slice of the data space and is made up of a collection
* of {@link org.onosproject.cluster.ControllerNode nodes}
* that all maintain copies of this data.
* A partition or shard is a group of controller nodes that are work together to maintain state.
* A ONOS cluster is typically made of of one or partitions over which the the data is partitioned.
*/
public class Partition {
private final String name;
private final Set<NodeId> members;
private Partition() {
name = null;
members = null;
}
public Partition(String name, Collection<NodeId> members) {
this.name = checkNotNull(name);
this.members = ImmutableSet.copyOf(checkNotNull(members));
}
public interface Partition {
/**
* Returns the partition name.
* <p>
* Each partition is identified by a unique name.
* @return partition name
* Returns the partition identifier.
* @return partition identifier
*/
public String getName() {
return this.name;
}
PartitionId getId();
/**
* Returns the collection of controller node identifiers that make up this partition.
* Returns the controller nodes that are members of this partition.
* @return collection of controller node identifiers
*/
public Collection<NodeId> getMembers() {
return this.members;
}
@Override
public int hashCode() {
return Arrays.deepHashCode(new Object[] {name, members});
}
@Override
public boolean equals(Object other) {
if (this == other) {
return true;
}
if (other == null || !Partition.class.isInstance(other)) {
return false;
}
Partition that = (Partition) other;
if (!this.name.equals(that.name) || (this.members == null && that.members != null)
|| (this.members != null && that.members == null) || this.members.size() != that.members.size()) {
return false;
}
return Sets.symmetricDifference(this.members, that.members).isEmpty();
}
}
\ No newline at end of file
Collection<NodeId> getMembers();
}
......
/*
* Copyright 2016 Open Networking Laboratory
*
* 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.onosproject.cluster;
import static com.google.common.base.Preconditions.checkArgument;
import java.util.Objects;
/**
* {@link Partition} identifier.
*/
public class PartitionId implements Comparable<PartitionId> {
private final int id;
/**
* Creates a partition identifier from an integer.
*
* @param id input integer
*/
public PartitionId(int id) {
checkArgument(id >= 0, "partition id must be non-negative");
this.id = id;
}
/**
* Creates a partition identifier from an integer.
*
* @param id input integer
*/
public static PartitionId from(int id) {
return new PartitionId(id);
}
/**
* Returns the partition identifier as an integer.
* @return number
*/
public int asInt() {
return id;
}
@Override
public int hashCode() {
return id;
}
@Override
public boolean equals(Object obj) {
if (this == obj) {
return true;
}
if (obj instanceof PartitionId) {
final PartitionId other = (PartitionId) obj;
return Objects.equals(this.id, other.id);
}
return false;
}
@Override
public String toString() {
return String.valueOf(id);
}
@Override
public int compareTo(PartitionId that) {
return Integer.compare(this.id, that.id);
}
}
\ No newline at end of file
......@@ -28,6 +28,7 @@ import org.onosproject.cluster.ClusterAdminService;
import org.onosproject.cluster.ClusterEvent;
import org.onosproject.cluster.ClusterEventListener;
import org.onosproject.cluster.ClusterMetadata;
import org.onosproject.cluster.ClusterMetadataAdminService;
import org.onosproject.cluster.ClusterMetadataService;
import org.onosproject.cluster.ClusterService;
import org.onosproject.cluster.ClusterStore;
......@@ -35,10 +36,13 @@ import org.onosproject.cluster.ClusterStoreDelegate;
import org.onosproject.cluster.ControllerNode;
import org.onosproject.cluster.NodeId;
import org.onosproject.cluster.Partition;
import org.onosproject.cluster.PartitionId;
import org.onosproject.event.AbstractListenerManager;
import org.slf4j.Logger;
import com.google.common.collect.Collections2;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import java.util.ArrayList;
import java.util.Collection;
......@@ -71,6 +75,9 @@ public class ClusterManager
protected ClusterMetadataService clusterMetadataService;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected ClusterMetadataAdminService clusterMetadataAdminService;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected ClusterStore store;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
......@@ -136,7 +143,7 @@ public class ClusterManager
.withControllerNodes(nodes)
.withPartitions(buildDefaultPartitions(nodes))
.build();
clusterMetadataService.setClusterMetadata(metadata);
clusterMetadataAdminService.setClusterMetadata(metadata);
try {
log.warn("Shutting down container for cluster reconfiguration!");
systemService.reboot("now", SystemService.Swipe.NONE);
......@@ -171,15 +178,36 @@ public class ClusterManager
List<ControllerNode> sorted = new ArrayList<>(nodes);
Collections.sort(sorted, (o1, o2) -> o1.id().toString().compareTo(o2.id().toString()));
Collection<Partition> partitions = Lists.newArrayList();
// add p0 partition
partitions.add(new Partition() {
@Override
public PartitionId getId() {
return PartitionId.from((0));
}
@Override
public Collection<NodeId> getMembers() {
return Sets.newHashSet(Collections2.transform(nodes, ControllerNode::id));
}
});
// add extended partitions
int length = nodes.size();
int count = 3;
for (int i = 0; i < length; i++) {
int index = i;
Set<NodeId> set = new HashSet<>(count);
for (int j = 0; j < count; j++) {
set.add(sorted.get((i + j) % length).id());
}
partitions.add(new Partition("p" + (i + 1), set));
partitions.add(new Partition() {
@Override
public PartitionId getId() {
return PartitionId.from((index + 1));
}
@Override
public Collection<NodeId> getMembers() {
return set;
}
});
}
return partitions;
}
......
/*
* Copyright 2015-2016 Open Networking Laboratory
*
* 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.onosproject.cluster.impl;
import static com.google.common.base.Preconditions.checkNotNull;
......@@ -17,6 +32,7 @@ import org.apache.felix.scr.annotations.ReferenceCardinality;
import org.apache.felix.scr.annotations.Service;
import org.onlab.packet.IpAddress;
import org.onosproject.cluster.ClusterMetadata;
import org.onosproject.cluster.ClusterMetadataAdminService;
import org.onosproject.cluster.ClusterMetadataEvent;
import org.onosproject.cluster.ClusterMetadataEventListener;
import org.onosproject.cluster.ClusterMetadataService;
......@@ -34,10 +50,10 @@ import org.slf4j.Logger;
@Service
public class ClusterMetadataManager
extends AbstractListenerManager<ClusterMetadataEvent, ClusterMetadataEventListener>
implements ClusterMetadataService {
implements ClusterMetadataService, ClusterMetadataAdminService {
private ControllerNode localNode;
private final Logger log = getLogger(getClass());
private ControllerNode localNode;
private ClusterMetadataStoreDelegate delegate = new InternalStoreDelegate();
......
/*
* Copyright 2015-2016 Open Networking Laboratory
*
* 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.onosproject.store.cluster.impl;
import static com.google.common.base.Preconditions.checkNotNull;
import java.util.Collection;
import java.util.Objects;
import org.onosproject.cluster.NodeId;
import org.onosproject.cluster.Partition;
import org.onosproject.cluster.PartitionId;
import com.google.common.base.MoreObjects;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Sets;
/**
* Default {@link Partition} implementation.
*/
public class DefaultPartition implements Partition {
private final PartitionId id;
private final Collection<NodeId> members;
private DefaultPartition() {
id = null;
members = null;
}
public DefaultPartition(PartitionId id, Collection<NodeId> members) {
this.id = checkNotNull(id);
this.members = ImmutableSet.copyOf(members);
}
@Override
public PartitionId getId() {
return this.id;
}
@Override
public Collection<NodeId> getMembers() {
return this.members;
}
@Override
public String toString() {
return MoreObjects.toStringHelper(getClass())
.add("id", id)
.add("members", members)
.toString();
}
@Override
public int hashCode() {
return Objects.hash(id, members);
}
@Override
public boolean equals(Object other) {
if (!(other instanceof DefaultPartition)) {
return false;
}
DefaultPartition that = (DefaultPartition) other;
return this.getId().equals(that.getId()) &&
Sets.symmetricDifference(Sets.newHashSet(this.members), Sets.newHashSet(that.members)).isEmpty();
}
}
\ No newline at end of file
/*
* Copyright 2015-2016 Open Networking Laboratory
*
* 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.onosproject.store.cluster.impl;
import static com.google.common.base.Preconditions.checkNotNull;
......@@ -28,6 +43,7 @@ import org.onosproject.cluster.ControllerNode;
import org.onosproject.cluster.DefaultControllerNode;
import org.onosproject.cluster.NodeId;
import org.onosproject.cluster.Partition;
import org.onosproject.cluster.PartitionId;
import org.onosproject.store.AbstractStore;
import org.onosproject.store.service.Versioned;
import org.slf4j.Logger;
......@@ -44,6 +60,7 @@ import com.fasterxml.jackson.databind.SerializerProvider;
import com.fasterxml.jackson.databind.module.SimpleModule;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.common.io.Files;
/**
......@@ -76,6 +93,7 @@ public class StaticClusterMetadataStore
module.addDeserializer(NodeId.class, new NodeIdDeserializer());
module.addSerializer(ControllerNode.class, new ControllerNodeSerializer());
module.addDeserializer(ControllerNode.class, new ControllerNodeDeserializer());
module.addDeserializer(Partition.class, new PartitionDeserializer());
mapper.registerModule(module);
File metadataFile = new File(CLUSTER_METADATA_FILE);
if (metadataFile.exists()) {
......@@ -89,10 +107,21 @@ public class StaticClusterMetadataStore
String localIp = getSiteLocalAddress();
ControllerNode localNode =
new DefaultControllerNode(new NodeId(localIp), IpAddress.valueOf(localIp), DEFAULT_ONOS_PORT);
Partition defaultPartition = new Partition() {
@Override
public PartitionId getId() {
return PartitionId.from(1);
}
@Override
public Collection<NodeId> getMembers() {
return Sets.newHashSet(localNode.id());
}
};
metadata.set(ClusterMetadata.builder()
.withName("default")
.withControllerNodes(Arrays.asList(localNode))
.withPartitions(Lists.newArrayList(new Partition("p1", Lists.newArrayList(localNode.id()))))
.withPartitions(Lists.newArrayList(defaultPartition))
.build());
version = System.currentTimeMillis();
}
......@@ -138,25 +167,33 @@ public class StaticClusterMetadataStore
}
@Override
public void setActiveReplica(String partitionId, NodeId nodeId) {
public void addActivePartitionMember(PartitionId partitionId, NodeId nodeId) {
throw new UnsupportedOperationException();
}
@Override
public void unsetActiveReplica(String partitionId, NodeId nodeId) {
public void removeActivePartitionMember(PartitionId partitionId, NodeId nodeId) {
throw new UnsupportedOperationException();
}
@Override
public Collection<NodeId> getActiveReplicas(String partitionId) {
public Collection<NodeId> getActivePartitionMembers(PartitionId partitionId) {
return metadata.get().getPartitions()
.stream()
.filter(r -> r.getName().equals(partitionId))
.filter(r -> r.getId().equals(partitionId))
.findFirst()
.map(r -> r.getMembers())
.orElse(null);
}
private static class PartitionDeserializer extends JsonDeserializer<Partition> {
@Override
public Partition deserialize(JsonParser jp, DeserializationContext ctxt)
throws IOException, JsonProcessingException {
return jp.readValueAs(DefaultPartition.class);
}
}
private static class ControllerNodeSerializer extends JsonSerializer<ControllerNode> {
@Override
public void serialize(ControllerNode node, JsonGenerator jgen, SerializerProvider provider)
......
......@@ -53,6 +53,7 @@ import org.onosproject.cluster.ClusterMetadataService;
import org.onosproject.cluster.ClusterService;
import org.onosproject.cluster.ControllerNode;
import org.onosproject.cluster.NodeId;
import org.onosproject.cluster.PartitionId;
import org.onosproject.core.ApplicationId;
import org.onosproject.core.IdGenerator;
import org.onosproject.persistence.PersistenceService;
......@@ -82,6 +83,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Function;
import java.util.stream.Collectors;
import static org.slf4j.LoggerFactory.getLogger;
......@@ -151,12 +153,11 @@ public class DatabaseManager implements StorageService, StorageAdminService {
public void activate() {
localNodeId = clusterService.getLocalNode().id();
Map<String, Set<NodeId>> partitionMap = Maps.newHashMap();
Map<PartitionId, Set<NodeId>> partitionMap = Maps.newHashMap();
clusterMetadataService.getClusterMetadata().getPartitions().forEach(p -> {
partitionMap.put(p.getName(), Sets.newHashSet(p.getMembers()));
partitionMap.put(p.getId(), Sets.newHashSet(p.getMembers()));
});
String[] activeNodeUris = partitionMap.values()
.stream()
.reduce((s1, s2) -> Sets.union(s1, s2))
......@@ -183,28 +184,19 @@ public class DatabaseManager implements StorageService, StorageAdminService {
coordinator = new DefaultClusterCoordinator(copycatConfig.resolve());
DatabaseConfig inMemoryDatabaseConfig =
newDatabaseConfig(BASE_PARTITION_NAME, newInMemoryLog(), activeNodeUris);
inMemoryDatabase = coordinator
.getResource(inMemoryDatabaseConfig.getName(), inMemoryDatabaseConfig.resolve(clusterConfig)
.withSerializer(copycatConfig.getDefaultSerializer())
.withDefaultExecutor(copycatConfig.getDefaultExecutor()));
Function<PartitionId, Log> logFunction = id -> id.asInt() == 0 ? newInMemoryLog() : newPersistentLog();
List<Database> partitions = partitionMap.entrySet()
.stream()
.map(entry -> {
String[] replicas = entry.getValue().stream().map(this::nodeIdToUri).toArray(String[]::new);
return newDatabaseConfig(entry.getKey(), newPersistentLog(), replicas);
})
.map(config -> {
Database db = coordinator.getResource(config.getName(), config.resolve(clusterConfig)
.withSerializer(copycatConfig.getDefaultSerializer())
.withDefaultExecutor(copycatConfig.getDefaultExecutor()));
return db;
})
.collect(Collectors.toList());
Map<PartitionId, Database> databases = Maps.transformEntries(partitionMap, (k, v) -> {
String[] replicas = v.stream().map(this::nodeIdToUri).toArray(String[]::new);
DatabaseConfig config = newDatabaseConfig(String.format("p%s", k), logFunction.apply(k), replicas);
return coordinator.getResource(config.getName(), config.resolve(clusterConfig)
.withSerializer(copycatConfig.getDefaultSerializer())
.withDefaultExecutor(copycatConfig.getDefaultExecutor()));
});
inMemoryDatabase = databases.remove(PartitionId.from(0));
partitionedDatabase = new PartitionedDatabase("onos-store", partitions);
partitionedDatabase = new PartitionedDatabase("onos-store", databases.values());
CompletableFuture<Void> status = coordinator.open()
.thenCompose(v -> CompletableFuture.allOf(inMemoryDatabase.open(), partitionedDatabase.open())
......
......@@ -32,11 +32,13 @@ import org.onlab.packet.IpAddress;
import org.onlab.packet.ONOSLLDP;
import org.onosproject.cfg.ComponentConfigAdapter;
import org.onosproject.cluster.ClusterMetadata;
import org.onosproject.cluster.ClusterMetadataEventListener;
import org.onosproject.cluster.ClusterMetadataService;
import org.onosproject.cluster.ControllerNode;
import org.onosproject.cluster.DefaultControllerNode;
import org.onosproject.cluster.NodeId;
import org.onosproject.cluster.Partition;
import org.onosproject.cluster.PartitionId;
import org.onosproject.cluster.RoleInfo;
import org.onosproject.core.ApplicationId;
import org.onosproject.core.CoreService;
......@@ -77,6 +79,7 @@ import org.onosproject.net.provider.AbstractProviderService;
import org.onosproject.net.provider.ProviderId;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
......@@ -950,7 +953,16 @@ public class LldpLinkProviderTest {
public ClusterMetadata getClusterMetadata() {
final NodeId nid = new NodeId("test-node");
final IpAddress addr = IpAddress.valueOf(0);
final Partition p = new Partition("test-pt", Sets.newHashSet(nid));
final Partition p = new Partition() {
public PartitionId getId() {
return PartitionId.from(1);
}
@Override
public Collection<NodeId> getMembers() {
return Sets.newHashSet(nid);
}
};
return ClusterMetadata.builder()
.withName("test-cluster")
.withControllerNodes(Sets.newHashSet(new DefaultControllerNode(nid, addr)))
......@@ -958,12 +970,16 @@ public class LldpLinkProviderTest {
}
@Override
public void setClusterMetadata(ClusterMetadata metadata) {
public ControllerNode getLocalNode() {
return null;
}
@Override
public void addListener(ClusterMetadataEventListener listener) {
}
@Override
public ControllerNode getLocalNode() {
return null;
public void removeListener(ClusterMetadataEventListener listener) {
}
}
}
......
......@@ -150,7 +150,7 @@ cat > $STAGE/config/cluster.json <<EOF
{
"name": "default",
"nodes": [ {"id": "$IP", "ip": "$IP", "port": 9876 } ],
"partitions": [ { "name": "p1", "members": [ "$IP" ] } ]
"partitions": [ { "id": 0, "members": [ "$IP" ] }, { "id": 1, "members": [ "$IP" ] } ]
}
EOF
......
......@@ -27,12 +27,18 @@ def get_nodes(vars, port=9876):
node = lambda k: { 'id': k, 'ip': k, 'port': port }
return [ node(environ[v]) for v in vars ]
def generate_permutations(nodes, k):
def generate_base_partition(nodes):
return {
'id': 0,
'members': nodes
}
def generate_extended_partitions(nodes, k):
l = deque(nodes)
perms = []
for i in range(1, len(nodes)+1):
part = {
'name': 'p%d' % i,
'id': i,
'members': list(l)[:k]
}
perms.append(part)
......@@ -42,10 +48,14 @@ def generate_permutations(nodes, k):
if __name__ == '__main__':
vars = get_OC_vars()
nodes = get_nodes(vars)
partitions = generate_permutations([v.get('id') for v in nodes], 3)
base_partition = generate_base_partition([v.get('id') for v in nodes])
extended_partitions = generate_extended_partitions([v.get('id') for v in nodes], 3)
partitions = []
partitions.append(base_partition)
partitions.extend(extended_partitions)
name = 0
for node in nodes:
name = name ^ hash(node['ip'])
name = name ^ hash(node['ip'])
data = {
'name': name,
'nodes': nodes,
......