blob: b78f3aedae6c6bc3dc28b2efc1b88397f1c0a3f6 [file] [log] [blame]
Madan Jampani25461112015-02-17 14:17:29 -08001/*
2 * Copyright 2015 Open Networking Laboratory
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 * http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
Madan Jampani09342702015-02-05 23:32:40 -080017package org.onosproject.store.consistent.impl;
18
Jonathan Hart054da972015-02-18 17:30:28 -080019import com.google.common.collect.Sets;
Madan Jampanid14166a2015-02-24 17:37:51 -080020
Madan Jampani09342702015-02-05 23:32:40 -080021import net.kuujo.copycat.cluster.ClusterConfig;
Jonathan Hart054da972015-02-18 17:30:28 -080022import net.kuujo.copycat.cluster.Member;
Madan Jampani09342702015-02-05 23:32:40 -080023import net.kuujo.copycat.log.FileLog;
24import net.kuujo.copycat.netty.NettyTcpProtocol;
25import net.kuujo.copycat.protocol.Consistency;
Madan Jampanid14166a2015-02-24 17:37:51 -080026
Madan Jampani09342702015-02-05 23:32:40 -080027import org.apache.felix.scr.annotations.Activate;
28import org.apache.felix.scr.annotations.Component;
29import org.apache.felix.scr.annotations.Deactivate;
30import org.apache.felix.scr.annotations.Reference;
31import org.apache.felix.scr.annotations.ReferenceCardinality;
32import org.apache.felix.scr.annotations.Service;
33import org.onosproject.cluster.ClusterService;
Madan Jampani0cb00672015-02-27 00:27:22 -080034import org.onosproject.store.cluster.impl.NodeInfo;
Madan Jampani393e0f02015-02-12 07:35:39 +053035import org.onosproject.store.service.ConsistentMap;
Jonathan Hart054da972015-02-18 17:30:28 -080036import org.onosproject.store.service.PartitionInfo;
Madan Jampani393e0f02015-02-12 07:35:39 +053037import org.onosproject.store.service.Serializer;
Jonathan Hart054da972015-02-18 17:30:28 -080038import org.onosproject.store.service.StorageAdminService;
Madan Jampani393e0f02015-02-12 07:35:39 +053039import org.onosproject.store.service.StorageService;
Madan Jampani64689552015-02-17 10:00:27 -080040import org.onosproject.store.service.TransactionContext;
Madan Jampani09342702015-02-05 23:32:40 -080041import org.slf4j.Logger;
42
Jonathan Hart054da972015-02-18 17:30:28 -080043import java.io.File;
44import java.io.IOException;
45import java.util.List;
46import java.util.Map;
47import java.util.Set;
Madan Jampanid14166a2015-02-24 17:37:51 -080048import java.util.concurrent.CountDownLatch;
49import java.util.concurrent.TimeUnit;
Jonathan Hart054da972015-02-18 17:30:28 -080050import java.util.stream.Collectors;
51
52import static org.slf4j.LoggerFactory.getLogger;
Madan Jampani09342702015-02-05 23:32:40 -080053
54/**
55 * Database manager.
56 */
57@Component(immediate = true, enabled = true)
58@Service
Jonathan Hart054da972015-02-18 17:30:28 -080059public class DatabaseManager implements StorageService, StorageAdminService {
Madan Jampani09342702015-02-05 23:32:40 -080060
61 private final Logger log = getLogger(getClass());
62 private PartitionedDatabase partitionedDatabase;
63 public static final int COPYCAT_TCP_PORT = 7238; // 7238 = RAFT
64 private static final String CONFIG_DIR = "../config";
65 private static final String PARTITION_DEFINITION_FILE = "tablets.json";
Madan Jampanid14166a2015-02-24 17:37:51 -080066 private static final int DATABASE_STARTUP_TIMEOUT_SEC = 60;
Madan Jampani09342702015-02-05 23:32:40 -080067
Madan Jampanife3a9a72015-03-13 16:32:26 -070068 private final PartitionedDatabaseConfig databaseConfig = new PartitionedDatabaseConfig();
69
Madan Jampani09342702015-02-05 23:32:40 -080070 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
71 protected ClusterService clusterService;
72
Madan Jampani0cb00672015-02-27 00:27:22 -080073 protected String nodeToUri(NodeInfo node) {
74 return String.format("tcp://%s:%d", node.getIp(), COPYCAT_TCP_PORT);
Madan Jampani09342702015-02-05 23:32:40 -080075 }
76
77 @Activate
78 public void activate() {
79
80 final String logDir = System.getProperty("karaf.data", "./data");
81
82 // load database configuration
83 File file = new File(CONFIG_DIR, PARTITION_DEFINITION_FILE);
84 log.info("Loading database definition: {}", file.getAbsolutePath());
85
Madan Jampani0cb00672015-02-27 00:27:22 -080086 Map<String, Set<NodeInfo>> partitionMap;
Madan Jampani09342702015-02-05 23:32:40 -080087 try {
Madan Jampani0cb00672015-02-27 00:27:22 -080088 DatabaseDefinitionStore databaseDef = new DatabaseDefinitionStore(file);
89 partitionMap = databaseDef.read().getPartitions();
Madan Jampani09342702015-02-05 23:32:40 -080090 } catch (IOException e) {
Madan Jampani09342702015-02-05 23:32:40 -080091 throw new IllegalStateException("Failed to load database config", e);
92 }
93
94 String[] activeNodeUris = partitionMap.values()
95 .stream()
96 .reduce((s1, s2) -> Sets.union(s1, s2))
97 .get()
98 .stream()
99 .map(this::nodeToUri)
100 .toArray(String[]::new);
101
Madan Jampani0cb00672015-02-27 00:27:22 -0800102 String localNodeUri = nodeToUri(NodeInfo.of(clusterService.getLocalNode()));
Madan Jampani09342702015-02-05 23:32:40 -0800103
104 ClusterConfig clusterConfig = new ClusterConfig()
Madan Jampani393e0f02015-02-12 07:35:39 +0530105 .withProtocol(new NettyTcpProtocol()
106 .withSsl(false)
107 .withConnectTimeout(60000)
108 .withAcceptBacklog(1024)
109 .withTrafficClass(-1)
110 .withSoLinger(-1)
111 .withReceiveBufferSize(32768)
112 .withSendBufferSize(8192)
113 .withThreads(1))
Thomas Vachuska0249b532015-02-20 16:46:18 -0800114 .withElectionTimeout(3000)
115 .withHeartbeatInterval(1500)
Madan Jampani09342702015-02-05 23:32:40 -0800116 .withMembers(activeNodeUris)
117 .withLocalMember(localNodeUri);
118
Madan Jampani09342702015-02-05 23:32:40 -0800119 partitionMap.forEach((name, nodes) -> {
120 Set<String> replicas = nodes.stream().map(this::nodeToUri).collect(Collectors.toSet());
121 DatabaseConfig partitionConfig = new DatabaseConfig()
Thomas Vachuska0249b532015-02-20 16:46:18 -0800122 .withElectionTimeout(3000)
123 .withHeartbeatInterval(1500)
Madan Jampani09342702015-02-05 23:32:40 -0800124 .withConsistency(Consistency.STRONG)
Madan Jampani393e0f02015-02-12 07:35:39 +0530125 .withLog(new FileLog()
126 .withDirectory(logDir)
127 .withSegmentSize(1073741824) // 1GB
128 .withFlushOnWrite(true)
129 .withSegmentInterval(Long.MAX_VALUE))
130 .withDefaultSerializer(new DatabaseSerializer())
Madan Jampani09342702015-02-05 23:32:40 -0800131 .withReplicas(replicas);
132 databaseConfig.addPartition(name, partitionConfig);
133 });
134
135 partitionedDatabase = PartitionedDatabaseManager.create("onos-store", clusterConfig, databaseConfig);
136
Madan Jampanid14166a2015-02-24 17:37:51 -0800137 CountDownLatch latch = new CountDownLatch(1);
Madan Jampani09342702015-02-05 23:32:40 -0800138 partitionedDatabase.open().whenComplete((db, error) -> {
139 if (error != null) {
140 log.warn("Failed to open database.", error);
141 } else {
Madan Jampanid14166a2015-02-24 17:37:51 -0800142 latch.countDown();
Madan Jampani09342702015-02-05 23:32:40 -0800143 log.info("Successfully opened database.");
144 }
145 });
Madan Jampanid14166a2015-02-24 17:37:51 -0800146 try {
147 if (!latch.await(DATABASE_STARTUP_TIMEOUT_SEC, TimeUnit.SECONDS)) {
Madan Jampani7f72c3f2015-03-01 17:34:59 -0800148 log.warn("Timed out waiting for database to initialize.");
Madan Jampanid14166a2015-02-24 17:37:51 -0800149 }
150 } catch (InterruptedException e) {
151 Thread.currentThread().interrupt();
152 log.warn("Failed to complete database initialization.");
153 }
Madan Jampani09342702015-02-05 23:32:40 -0800154 log.info("Started");
155 }
156
157 @Deactivate
158 public void deactivate() {
159 partitionedDatabase.close().whenComplete((result, error) -> {
160 if (error != null) {
161 log.warn("Failed to cleanly close database.", error);
162 } else {
163 log.info("Successfully closed database.");
164 }
165 });
166 log.info("Stopped");
167 }
168
169 @Override
Madan Jampani393e0f02015-02-12 07:35:39 +0530170 public <K, V> ConsistentMap<K , V> createConsistentMap(String name, Serializer serializer) {
Madan Jampani09342702015-02-05 23:32:40 -0800171 return new ConsistentMapImpl<K, V>(name, partitionedDatabase, serializer);
172 }
Madan Jampani64689552015-02-17 10:00:27 -0800173
174 @Override
175 public TransactionContext createTransactionContext() {
176 return new DefaultTransactionContext(partitionedDatabase);
177 }
Jonathan Hart054da972015-02-18 17:30:28 -0800178
179 @Override
180 public List<PartitionInfo> getPartitionInfo() {
181 return partitionedDatabase.getRegisteredPartitions()
182 .values()
183 .stream()
Madan Jampanife3a9a72015-03-13 16:32:26 -0700184 .map(db -> toPartitionInfo(db, databaseConfig.partitions().get(db.name())))
Jonathan Hart054da972015-02-18 17:30:28 -0800185 .collect(Collectors.toList());
186 }
187
188 /**
189 * Maps a Raft Database object to a PartitionInfo object.
190 *
191 * @param database database containing input data
192 * @return PartitionInfo object
193 */
Madan Jampanife3a9a72015-03-13 16:32:26 -0700194 private static PartitionInfo toPartitionInfo(Database database, DatabaseConfig dbConfig) {
Jonathan Hart054da972015-02-18 17:30:28 -0800195 return new PartitionInfo(database.name(),
196 database.cluster().term(),
197 database.cluster().members().stream()
198 .map(Member::uri)
Madan Jampanife3a9a72015-03-13 16:32:26 -0700199 .filter(uri -> dbConfig.getReplicas().contains(uri))
Jonathan Hart054da972015-02-18 17:30:28 -0800200 .collect(Collectors.toList()),
201 database.cluster().leader() != null ?
202 database.cluster().leader().uri() : null);
203 }
204}