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
diff --git a/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/DefaultPartition.java b/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/DefaultPartition.java
new file mode 100644
index 0000000..82a3ba7
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/DefaultPartition.java
@@ -0,0 +1,81 @@
+/*
+ * 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
diff --git a/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/StaticClusterMetadataStore.java b/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/StaticClusterMetadataStore.java
index 3cd992b..da4ec99 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/StaticClusterMetadataStore.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/cluster/impl/StaticClusterMetadataStore.java
@@ -1,3 +1,18 @@
+/*
+ * 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.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.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 @@
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 @@
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 @@
}
@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)
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseManager.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseManager.java
index 90d81ee..8436526 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseManager.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseManager.java
@@ -53,6 +53,7 @@
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.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 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 @@
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()));
+ });
- partitionedDatabase = new PartitionedDatabase("onos-store", partitions);
+ inMemoryDatabase = databases.remove(PartitionId.from(0));
+
+ partitionedDatabase = new PartitionedDatabase("onos-store", databases.values());
CompletableFuture<Void> status = coordinator.open()
.thenCompose(v -> CompletableFuture.allOf(inMemoryDatabase.open(), partitionedDatabase.open())