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