gerrit.opnfv Code Review - onosfw.git/blob

   1 /*
   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
  17 package org.onosproject.store.consistent.impl;
  18
  19 import com.google.common.collect.ArrayListMultimap;
  20 import com.google.common.collect.ImmutableList;
  21 import com.google.common.collect.ImmutableSet;
  22 import com.google.common.collect.Lists;
  23 import com.google.common.collect.Maps;
  24 import com.google.common.collect.Multimap;
  25 import com.google.common.collect.Multimaps;
  26 import com.google.common.collect.Sets;
  27 import com.google.common.util.concurrent.Futures;
  28
  29 import net.kuujo.copycat.CopycatConfig;
  30 import net.kuujo.copycat.cluster.ClusterConfig;
  31 import net.kuujo.copycat.cluster.Member;
  32 import net.kuujo.copycat.cluster.Member.Type;
  33 import net.kuujo.copycat.cluster.internal.coordinator.ClusterCoordinator;
  34 import net.kuujo.copycat.cluster.internal.coordinator.DefaultClusterCoordinator;
  35 import net.kuujo.copycat.log.BufferedLog;
  36 import net.kuujo.copycat.log.FileLog;
  37 import net.kuujo.copycat.log.Log;
  38 import net.kuujo.copycat.protocol.Consistency;
  39 import net.kuujo.copycat.protocol.Protocol;
  40 import net.kuujo.copycat.util.concurrent.NamedThreadFactory;
  41
  42 import org.apache.commons.lang.math.RandomUtils;
  43 import org.apache.felix.scr.annotations.Activate;
  44 import org.apache.felix.scr.annotations.Component;
  45 import org.apache.felix.scr.annotations.Deactivate;
  46 import org.apache.felix.scr.annotations.Reference;
  47 import org.apache.felix.scr.annotations.ReferenceCardinality;
  48 import org.apache.felix.scr.annotations.ReferencePolicy;
  49 import org.apache.felix.scr.annotations.Service;
  50
  51 import org.onosproject.app.ApplicationEvent;
  52 import org.onosproject.app.ApplicationListener;
  53 import org.onosproject.app.ApplicationService;
  54 import org.onosproject.cluster.ClusterService;
  55 import org.onosproject.cluster.NodeId;
  56 import org.onosproject.core.ApplicationId;
  57 import org.onosproject.core.IdGenerator;
  58 import org.onosproject.store.cluster.impl.ClusterDefinitionManager;
  59 import org.onosproject.store.cluster.impl.NodeInfo;
  60 import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
  61 import org.onosproject.store.ecmap.EventuallyConsistentMapBuilderImpl;
  62 import org.onosproject.store.service.AtomicCounterBuilder;
  63 import org.onosproject.store.service.AtomicValueBuilder;
  64 import org.onosproject.store.service.ConsistentMapBuilder;
  65 import org.onosproject.store.service.ConsistentMapException;
  66 import org.onosproject.store.service.DistributedQueueBuilder;
  67 import org.onosproject.store.service.EventuallyConsistentMapBuilder;
  68 import org.onosproject.store.service.MapInfo;
  69 import org.onosproject.store.service.PartitionInfo;
  70 import org.onosproject.store.service.DistributedSetBuilder;
  71 import org.onosproject.store.service.StorageAdminService;
  72 import org.onosproject.store.service.StorageService;
  73 import org.onosproject.store.service.Transaction;
  74 import org.onosproject.store.service.TransactionContextBuilder;
  75 import org.slf4j.Logger;
  76
  77 import java.io.File;
  78 import java.io.IOException;
  79 import java.util.Collection;
  80 import java.util.List;
  81 import java.util.Map;
  82 import java.util.Set;
  83 import java.util.concurrent.CompletableFuture;
  84 import java.util.concurrent.ExecutionException;
  85 import java.util.concurrent.Executors;
  86 import java.util.concurrent.TimeUnit;
  87 import java.util.concurrent.TimeoutException;
  88 import java.util.stream.Collectors;
  89
  90 import static org.slf4j.LoggerFactory.getLogger;
  91 import static org.onosproject.app.ApplicationEvent.Type.APP_UNINSTALLED;
  92 import static org.onosproject.app.ApplicationEvent.Type.APP_DEACTIVATED;
  93
  94 /**
  95  * Database manager.
  96  */
  97 @Component(immediate = true, enabled = true)
  98 @Service
  99 public class DatabaseManager implements StorageService, StorageAdminService {
 100
 101     private final Logger log = getLogger(getClass());
 102
 103     public static final int COPYCAT_TCP_PORT = 9876;
 104     public static final String PARTITION_DEFINITION_FILE = "../config/tablets.json";
 105     public static final String BASE_PARTITION_NAME = "p0";
 106
 107     private static final int RAFT_ELECTION_TIMEOUT_MILLIS = 3000;
 108     private static final int DATABASE_OPERATION_TIMEOUT_MILLIS = 5000;
 109
 110     private ClusterCoordinator coordinator;
 111     protected PartitionedDatabase partitionedDatabase;
 112     protected Database inMemoryDatabase;
 113     protected NodeId localNodeId;
 114
 115     private TransactionManager transactionManager;
 116     private final IdGenerator transactionIdGenerator = () -> RandomUtils.nextLong();
 117
 118     private ApplicationListener appListener = new InternalApplicationListener();
 119
 120     private final Multimap<String, DefaultAsyncConsistentMap> maps =
 121             Multimaps.synchronizedMultimap(ArrayListMultimap.create());
 122     private final Multimap<ApplicationId, DefaultAsyncConsistentMap> mapsByApplication =
 123             Multimaps.synchronizedMultimap(ArrayListMultimap.create());
 124
 125     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
 126     protected ClusterService clusterService;
 127
 128     @Reference(cardinality = ReferenceCardinality.OPTIONAL_UNARY, policy = ReferencePolicy.DYNAMIC)
 129     protected ApplicationService applicationService;
 130
 131     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
 132     protected ClusterCommunicationService clusterCommunicator;
 133
 134     protected String nodeToUri(NodeInfo node) {
 135         return String.format("onos://%s:%d", node.getIp(), node.getTcpPort());
 136     }
 137
 138     protected void bindApplicationService(ApplicationService service) {
 139         applicationService = service;
 140         applicationService.addListener(appListener);
 141     }
 142
 143     protected void unbindApplicationService(ApplicationService service) {
 144         applicationService.removeListener(appListener);
 145         this.applicationService = null;
 146     }
 147
 148     @Activate
 149     public void activate() {
 150         localNodeId = clusterService.getLocalNode().id();
 151         // load database configuration
 152         File databaseDefFile = new File(PARTITION_DEFINITION_FILE);
 153         log.info("Loading database definition: {}", databaseDefFile.getAbsolutePath());
 154
 155         Map<String, Set<NodeInfo>> partitionMap;
 156         try {
 157             DatabaseDefinitionStore databaseDefStore = new DatabaseDefinitionStore(databaseDefFile);
 158             if (!databaseDefFile.exists()) {
 159                 createDefaultDatabaseDefinition(databaseDefStore);
 160             }
 161             partitionMap = databaseDefStore.read().getPartitions();
 162         } catch (IOException e) {
 163             throw new IllegalStateException("Failed to load database config", e);
 164         }
 165
 166         String[] activeNodeUris = partitionMap.values()
 167                     .stream()
 168                     .reduce((s1, s2) -> Sets.union(s1, s2))
 169                     .get()
 170                     .stream()
 171                     .map(this::nodeToUri)
 172                     .toArray(String[]::new);
 173
 174         String localNodeUri = nodeToUri(NodeInfo.of(clusterService.getLocalNode()));
 175         Protocol protocol = new CopycatCommunicationProtocol(clusterService, clusterCommunicator);
 176
 177         ClusterConfig clusterConfig = new ClusterConfig()
 178             .withProtocol(protocol)
 179             .withElectionTimeout(electionTimeoutMillis(activeNodeUris))
 180             .withHeartbeatInterval(heartbeatTimeoutMillis(activeNodeUris))
 181             .withMembers(activeNodeUris)
 182             .withLocalMember(localNodeUri);
 183
 184         CopycatConfig copycatConfig = new CopycatConfig()
 185             .withName("onos")
 186             .withClusterConfig(clusterConfig)
 187             .withDefaultSerializer(new DatabaseSerializer())
 188             .withDefaultExecutor(Executors.newSingleThreadExecutor(new NamedThreadFactory("copycat-coordinator-%d")));
 189
 190         coordinator = new DefaultClusterCoordinator(copycatConfig.resolve());
 191
 192         DatabaseConfig inMemoryDatabaseConfig =
 193                 newDatabaseConfig(BASE_PARTITION_NAME, newInMemoryLog(), activeNodeUris);
 194         inMemoryDatabase = coordinator
 195                 .getResource(inMemoryDatabaseConfig.getName(), inMemoryDatabaseConfig.resolve(clusterConfig)
 196                 .withSerializer(copycatConfig.getDefaultSerializer())
 197                 .withDefaultExecutor(copycatConfig.getDefaultExecutor()));
 198
 199         List<Database> partitions = partitionMap.entrySet()
 200             .stream()
 201             .map(entry -> {
 202                 String[] replicas = entry.getValue().stream().map(this::nodeToUri).toArray(String[]::new);
 203                 return newDatabaseConfig(entry.getKey(), newPersistentLog(), replicas);
 204                 })
 205             .map(config -> {
 206                 Database db = coordinator.getResource(config.getName(), config.resolve(clusterConfig)
 207                         .withSerializer(copycatConfig.getDefaultSerializer())
 208                         .withDefaultExecutor(copycatConfig.getDefaultExecutor()));
 209                 return db;
 210             })
 211             .collect(Collectors.toList());
 212
 213         partitionedDatabase = new PartitionedDatabase("onos-store", partitions);
 214
 215         CompletableFuture<Void> status = coordinator.open()
 216             .thenCompose(v -> CompletableFuture.allOf(inMemoryDatabase.open(), partitionedDatabase.open())
 217             .whenComplete((db, error) -> {
 218                 if (error != null) {
 219                     log.error("Failed to initialize database.", error);
 220                 } else {
 221                     log.info("Successfully initialized database.");
 222                 }
 223             }));
 224
 225         Futures.getUnchecked(status);
 226
 227         transactionManager = new TransactionManager(partitionedDatabase, consistentMapBuilder());
 228         partitionedDatabase.setTransactionManager(transactionManager);
 229
 230         log.info("Started");
 231     }
 232
 233     private void createDefaultDatabaseDefinition(DatabaseDefinitionStore store) {
 234         // Assumes IPv4 is returned.
 235         String ip = ClusterDefinitionManager.getSiteLocalAddress();
 236         NodeInfo node = NodeInfo.from(ip, ip, COPYCAT_TCP_PORT);
 237         try {
 238             store.write(DatabaseDefinition.from(ImmutableSet.of(node)));
 239         } catch (IOException e) {
 240             log.warn("Unable to write default cluster definition", e);
 241         }
 242     }
 243
 244     @Deactivate
 245     public void deactivate() {
 246         CompletableFuture.allOf(inMemoryDatabase.close(), partitionedDatabase.close())
 247             .thenCompose(v -> coordinator.close())
 248             .whenComplete((result, error) -> {
 249                 if (error != null) {
 250                     log.warn("Failed to cleanly close databases.", error);
 251                 } else {
 252                     log.info("Successfully closed databases.");
 253                 }
 254             });
 255         maps.values().forEach(this::unregisterMap);
 256         if (applicationService != null) {
 257             applicationService.removeListener(appListener);
 258         }
 259         log.info("Stopped");
 260     }
 261
 262     @Override
 263     public TransactionContextBuilder transactionContextBuilder() {
 264         return new DefaultTransactionContextBuilder(this, transactionIdGenerator.getNewId());
 265     }
 266
 267     @Override
 268     public List<PartitionInfo> getPartitionInfo() {
 269         return Lists.asList(
 270                     inMemoryDatabase,
 271                     partitionedDatabase.getPartitions().toArray(new Database[]{}))
 272                 .stream()
 273                 .map(DatabaseManager::toPartitionInfo)
 274                 .collect(Collectors.toList());
 275     }
 276
 277     private Log newPersistentLog() {
 278         String logDir = System.getProperty("karaf.data", "./data");
 279         return new FileLog()
 280             .withDirectory(logDir)
 281             .withSegmentSize(1073741824) // 1GB
 282             .withFlushOnWrite(true)
 283             .withSegmentInterval(Long.MAX_VALUE);
 284     }
 285
 286     private Log newInMemoryLog() {
 287         return new BufferedLog()
 288             .withFlushOnWrite(false)
 289             .withFlushInterval(Long.MAX_VALUE)
 290             .withSegmentSize(10485760) // 10MB
 291             .withSegmentInterval(Long.MAX_VALUE);
 292     }
 293
 294     private DatabaseConfig newDatabaseConfig(String name, Log log, String[] replicas) {
 295         return new DatabaseConfig()
 296             .withName(name)
 297             .withElectionTimeout(electionTimeoutMillis(replicas))
 298             .withHeartbeatInterval(heartbeatTimeoutMillis(replicas))
 299             .withConsistency(Consistency.DEFAULT)
 300             .withLog(log)
 301             .withDefaultSerializer(new DatabaseSerializer())
 302             .withReplicas(replicas);
 303     }
 304
 305     private long electionTimeoutMillis(String[] replicas) {
 306         return replicas.length == 1 ? 10L : RAFT_ELECTION_TIMEOUT_MILLIS;
 307     }
 308
 309     private long heartbeatTimeoutMillis(String[] replicas) {
 310         return electionTimeoutMillis(replicas) / 2;
 311     }
 312
 313     /**
 314      * Maps a Raft Database object to a PartitionInfo object.
 315      *
 316      * @param database database containing input data
 317      * @return PartitionInfo object
 318      */
 319     private static PartitionInfo toPartitionInfo(Database database) {
 320         return new PartitionInfo(database.name(),
 321                           database.cluster().term(),
 322                           database.cluster().members()
 323                                   .stream()
 324                                   .filter(member -> Type.ACTIVE.equals(member.type()))
 325                                   .map(Member::uri)
 326                                   .sorted()
 327                                   .collect(Collectors.toList()),
 328                           database.cluster().leader() != null ?
 329                                   database.cluster().leader().uri() : null);
 330     }
 331
 332
 333     @Override
 334     public <K, V> EventuallyConsistentMapBuilder<K, V> eventuallyConsistentMapBuilder() {
 335         return new EventuallyConsistentMapBuilderImpl<>(clusterService,
 336                                                         clusterCommunicator);
 337     }
 338
 339     @Override
 340     public <K, V> ConsistentMapBuilder<K, V> consistentMapBuilder() {
 341         return new DefaultConsistentMapBuilder<>(this);
 342     }
 343
 344     @Override
 345     public <E> DistributedSetBuilder<E> setBuilder() {
 346         return new DefaultDistributedSetBuilder<>(this);
 347     }
 348
 349
 350     @Override
 351     public <E> DistributedQueueBuilder<E> queueBuilder() {
 352         return new DefaultDistributedQueueBuilder<>(this);
 353     }
 354
 355     @Override
 356     public AtomicCounterBuilder atomicCounterBuilder() {
 357         return new DefaultAtomicCounterBuilder(inMemoryDatabase, partitionedDatabase);
 358     }
 359
 360     @Override
 361     public <V> AtomicValueBuilder<V> atomicValueBuilder() {
 362         return new DefaultAtomicValueBuilder<>(this);
 363     }
 364
 365     @Override
 366     public List<MapInfo> getMapInfo() {
 367         List<MapInfo> maps = Lists.newArrayList();
 368         maps.addAll(getMapInfo(inMemoryDatabase));
 369         maps.addAll(getMapInfo(partitionedDatabase));
 370         return maps;
 371     }
 372
 373     private List<MapInfo> getMapInfo(Database database) {
 374         return complete(database.maps())
 375             .stream()
 376             .map(name -> new MapInfo(name, complete(database.mapSize(name))))
 377             .filter(info -> info.size() > 0)
 378             .collect(Collectors.toList());
 379     }
 380
 381
 382     @Override
 383     public Map<String, Long> getCounters() {
 384         Map<String, Long> counters = Maps.newHashMap();
 385         counters.putAll(complete(inMemoryDatabase.counters()));
 386         counters.putAll(complete(partitionedDatabase.counters()));
 387         return counters;
 388     }
 389
 390     @Override
 391     public Map<String, Long> getPartitionedDatabaseCounters() {
 392         Map<String, Long> counters = Maps.newHashMap();
 393         counters.putAll(complete(partitionedDatabase.counters()));
 394         return counters;
 395     }
 396
 397     @Override
 398     public Map<String, Long> getInMemoryDatabaseCounters() {
 399         Map<String, Long> counters = Maps.newHashMap();
 400         counters.putAll(complete(inMemoryDatabase.counters()));
 401         return counters;
 402     }
 403
 404     @Override
 405     public Collection<Transaction> getTransactions() {
 406         return complete(transactionManager.getTransactions());
 407     }
 408
 409     private static <T> T complete(CompletableFuture<T> future) {
 410         try {
 411             return future.get(DATABASE_OPERATION_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
 412         } catch (InterruptedException e) {
 413             Thread.currentThread().interrupt();
 414             throw new ConsistentMapException.Interrupted();
 415         } catch (TimeoutException e) {
 416             throw new ConsistentMapException.Timeout();
 417         } catch (ExecutionException e) {
 418             throw new ConsistentMapException(e.getCause());
 419         }
 420     }
 421
 422     @Override
 423     public void redriveTransactions() {
 424         getTransactions().stream().forEach(transactionManager::execute);
 425     }
 426
 427     protected <K, V> DefaultAsyncConsistentMap<K, V> registerMap(DefaultAsyncConsistentMap<K, V> map) {
 428         maps.put(map.name(), map);
 429         if (map.applicationId() != null) {
 430             mapsByApplication.put(map.applicationId(), map);
 431         }
 432         return map;
 433     }
 434
 435     protected <K, V> void unregisterMap(DefaultAsyncConsistentMap<K, V> map) {
 436         maps.remove(map.name(), map);
 437         if (map.applicationId() != null) {
 438             mapsByApplication.remove(map.applicationId(), map);
 439         }
 440     }
 441
 442     private class InternalApplicationListener implements ApplicationListener {
 443         @Override
 444         public void event(ApplicationEvent event) {
 445             if (event.type() == APP_UNINSTALLED || event.type() == APP_DEACTIVATED) {
 446                 ApplicationId appId = event.subject().id();
 447                 List<DefaultAsyncConsistentMap> mapsToRemove = ImmutableList.copyOf(mapsByApplication.get(appId));
 448                 mapsToRemove.forEach(DatabaseManager.this::unregisterMap);
 449                 if (event.type() == APP_UNINSTALLED) {
 450                     mapsToRemove.stream().filter(map -> map.purgeOnUninstall()).forEach(map -> map.clear());
 451                 }
 452             }
 453         }
 454     }
 455 }