Detangling incubator: virtual nets, tunnels, resource labels, oh my
- virtual networking moved to /apps/virtual; with CLI & REST API
- tunnels and labels moved to /apps/tunnel; with CLI & REST API; UI disabled for now
- protobuf/models moved to /core/protobuf/models
- defunct grpc/rpc registry stuff left under /graveyard
- compile dependencies on /incubator moved to respective modules for compilation
- run-time dependencies will need to be re-tested for dependent apps
- /graveyard will be removed in not-too-distant future
Change-Id: I0a0b995c635487edcf95a352f50dd162186b0b39
diff --git a/core/store/dist/src/main/java/org/onosproject/store/intent/impl/GossipIntentStore.java b/core/store/dist/src/main/java/org/onosproject/store/intent/impl/GossipIntentStore.java
index bddb7a0..9a0584a 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/intent/impl/GossipIntentStore.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/intent/impl/GossipIntentStore.java
@@ -23,8 +23,6 @@
import org.onosproject.cluster.ClusterService;
import org.onosproject.cluster.ControllerNode;
import org.onosproject.cluster.NodeId;
-import org.onosproject.incubator.net.virtual.NetworkId;
-import org.onosproject.incubator.net.virtual.VirtualNetworkIntent;
import org.onosproject.net.intent.Intent;
import org.onosproject.net.intent.IntentData;
import org.onosproject.net.intent.IntentEvent;
@@ -68,6 +66,7 @@
import static org.onosproject.store.OsgiPropertyConstants.GIS_PERSISTENCE_ENABLED;
import static org.onosproject.store.OsgiPropertyConstants.GIS_PERSISTENCE_ENABLED_DEFAULT;
import static org.slf4j.LoggerFactory.getLogger;
+
/**
* Manages inventory of Intents in a distributed data store that uses optimistic
* replication and gossip based techniques.
@@ -158,8 +157,6 @@
KryoNamespace.Builder intentSerializer = KryoNamespace.newBuilder()
.register(KryoNamespaces.API)
.register(IntentData.class)
- .register(VirtualNetworkIntent.class)
- .register(NetworkId.class)
.register(MultiValuedTimestamp.class);
EventuallyConsistentMapBuilder currentECMapBuilder =
diff --git a/core/store/dist/src/main/java/org/onosproject/store/meter/impl/DistributedMeterStore.java b/core/store/dist/src/main/java/org/onosproject/store/meter/impl/DistributedMeterStore.java
new file mode 100644
index 0000000..09a55ad
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onosproject/store/meter/impl/DistributedMeterStore.java
@@ -0,0 +1,500 @@
+/*
+ * Copyright 2015-present Open Networking Foundation
+ *
+ * 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.meter.impl;
+
+import com.google.common.collect.Collections2;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.apache.commons.lang.math.RandomUtils;
+import org.onlab.util.KryoNamespace;
+import org.onosproject.cluster.ClusterService;
+import org.onosproject.cluster.NodeId;
+import org.onosproject.mastership.MastershipService;
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.behaviour.MeterQuery;
+import org.onosproject.net.driver.DriverHandler;
+import org.onosproject.net.driver.DriverService;
+import org.onosproject.net.meter.Band;
+import org.onosproject.net.meter.DefaultBand;
+import org.onosproject.net.meter.DefaultMeter;
+import org.onosproject.net.meter.DefaultMeterFeatures;
+import org.onosproject.net.meter.Meter;
+import org.onosproject.net.meter.MeterEvent;
+import org.onosproject.net.meter.MeterFailReason;
+import org.onosproject.net.meter.MeterFeatures;
+import org.onosproject.net.meter.MeterFeaturesFlag;
+import org.onosproject.net.meter.MeterFeaturesKey;
+import org.onosproject.net.meter.MeterId;
+import org.onosproject.net.meter.MeterKey;
+import org.onosproject.net.meter.MeterOperation;
+import org.onosproject.net.meter.MeterState;
+import org.onosproject.net.meter.MeterStore;
+import org.onosproject.net.meter.MeterStoreDelegate;
+import org.onosproject.net.meter.MeterStoreResult;
+import org.onosproject.store.AbstractStore;
+import org.onosproject.store.primitives.DefaultDistributedSet;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.AtomicCounterMap;
+import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.DistributedPrimitive;
+import org.onosproject.store.service.DistributedSet;
+import org.onosproject.store.service.MapEvent;
+import org.onosproject.store.service.MapEventListener;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.StorageException;
+import org.onosproject.store.service.StorageService;
+import org.onosproject.store.service.Versioned;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.component.annotations.ReferenceCardinality;
+import org.slf4j.Logger;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+import static org.onosproject.store.meter.impl.DistributedMeterStore.ReuseStrategy.FIRST_FIT;
+import static org.onosproject.net.meter.MeterFailReason.TIMEOUT;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * A distributed meter store implementation. Meters are stored consistently
+ * across the cluster.
+ */
+@Component(immediate = true, service = MeterStore.class)
+public class DistributedMeterStore extends AbstractStore<MeterEvent, MeterStoreDelegate>
+ implements MeterStore {
+
+ private Logger log = getLogger(getClass());
+
+ private static final String METERSTORE = "onos-meter-store";
+ private static final String METERFEATURESSTORE = "onos-meter-features-store";
+ private static final String AVAILABLEMETERIDSTORE = "onos-meters-available-store";
+ private static final String METERIDSTORE = "onos-meters-id-store";
+
+ private static final KryoNamespace.Builder APP_KRYO_BUILDER = KryoNamespace.newBuilder()
+ .register(KryoNamespaces.API)
+ .register(MeterKey.class)
+ .register(MeterData.class)
+ .register(DefaultMeter.class)
+ .register(DefaultBand.class)
+ .register(Band.Type.class)
+ .register(MeterState.class)
+ .register(Meter.Unit.class);
+
+ private Serializer serializer = Serializer.using(Lists.newArrayList(APP_KRYO_BUILDER.build()));
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ private StorageService storageService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ private MastershipService mastershipService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ private ClusterService clusterService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ protected DriverService driverService;
+
+ private ConsistentMap<MeterKey, MeterData> meters;
+ private NodeId local;
+
+ private ConsistentMap<MeterFeaturesKey, MeterFeatures> meterFeatures;
+
+ private MapEventListener<MeterKey, MeterData> mapListener = new InternalMapEventListener();
+
+ private Map<MeterKey, CompletableFuture<MeterStoreResult>> futures =
+ Maps.newConcurrentMap();
+
+ // Available meter identifiers
+ private DistributedSet<MeterKey> availableMeterIds;
+
+ // Atomic counter map for generation of new identifiers;
+ private AtomicCounterMap<DeviceId> meterIdGenerators;
+
+ /**
+ * Defines possible selection strategies to reuse meter ids.
+ */
+ enum ReuseStrategy {
+ /**
+ * Select randomly an available id.
+ */
+ RANDOM,
+ /**
+ * Select the first one.
+ */
+ FIRST_FIT
+ }
+
+ private ReuseStrategy reuseStrategy = FIRST_FIT;
+
+ @Activate
+ public void activate() {
+ local = clusterService.getLocalNode().id();
+
+ meters = storageService.<MeterKey, MeterData>consistentMapBuilder()
+ .withName(METERSTORE)
+ .withSerializer(serializer).build();
+
+ meters.addListener(mapListener);
+
+ meterFeatures = storageService.<MeterFeaturesKey, MeterFeatures>consistentMapBuilder()
+ .withName(METERFEATURESSTORE)
+ .withSerializer(Serializer.using(KryoNamespaces.API,
+ MeterFeaturesKey.class,
+ MeterFeatures.class,
+ DefaultMeterFeatures.class,
+ Band.Type.class,
+ Meter.Unit.class,
+ MeterFailReason.class,
+ MeterFeaturesFlag.class)).build();
+
+ // Init the set of the available ids
+ availableMeterIds = new DefaultDistributedSet<>(storageService.<MeterKey>setBuilder()
+ .withName(AVAILABLEMETERIDSTORE)
+ .withSerializer(Serializer.using(KryoNamespaces.API,
+ MeterKey.class)).build(),
+ DistributedPrimitive.DEFAULT_OPERATION_TIMEOUT_MILLIS);
+
+ // Init atomic map counters
+ meterIdGenerators = storageService.<DeviceId>atomicCounterMapBuilder()
+ .withName(METERIDSTORE)
+ .withSerializer(Serializer.using(KryoNamespaces.API)).build();
+
+ log.info("Started");
+ }
+
+ @Deactivate
+ public void deactivate() {
+ meters.removeListener(mapListener);
+ log.info("Stopped");
+ }
+
+ @Override
+ public CompletableFuture<MeterStoreResult> storeMeter(Meter meter) {
+ // Init steps
+ CompletableFuture<MeterStoreResult> future = new CompletableFuture<>();
+ MeterKey key = MeterKey.key(meter.deviceId(), meter.id());
+ // Store the future related to the operation
+ futures.put(key, future);
+ // Store the meter data
+ MeterData data = new MeterData(meter, null, local);
+ try {
+ meters.put(key, data);
+ } catch (StorageException e) {
+ futures.remove(key);
+ future.completeExceptionally(e);
+ }
+ // Done, return the future
+ return future;
+ }
+
+ @Override
+ public CompletableFuture<MeterStoreResult> deleteMeter(Meter meter) {
+ // Init steps
+ CompletableFuture<MeterStoreResult> future = new CompletableFuture<>();
+ MeterKey key = MeterKey.key(meter.deviceId(), meter.id());
+ // Store the future related to the operation
+ futures.put(key, future);
+ // Create the meter data
+ MeterData data = new MeterData(meter, null, local);
+ // Update the state of the meter. It will be pruned by observing
+ // that it has been removed from the dataplane.
+ try {
+ // If it does not exist in the system
+ if (meters.computeIfPresent(key, (k, v) -> data) == null) {
+ // Complete immediately
+ future.complete(MeterStoreResult.success());
+ }
+ } catch (StorageException e) {
+ futures.remove(key);
+ future.completeExceptionally(e);
+ }
+ // Done, return the future
+ return future;
+ }
+
+ @Override
+ public MeterStoreResult storeMeterFeatures(MeterFeatures meterfeatures) {
+ MeterStoreResult result = MeterStoreResult.success();
+ MeterFeaturesKey key = MeterFeaturesKey.key(meterfeatures.deviceId());
+ try {
+ meterFeatures.putIfAbsent(key, meterfeatures);
+ } catch (StorageException e) {
+ result = MeterStoreResult.fail(TIMEOUT);
+ }
+ return result;
+ }
+
+ @Override
+ public MeterStoreResult deleteMeterFeatures(DeviceId deviceId) {
+ MeterStoreResult result = MeterStoreResult.success();
+ MeterFeaturesKey key = MeterFeaturesKey.key(deviceId);
+ try {
+ meterFeatures.remove(key);
+ } catch (StorageException e) {
+ result = MeterStoreResult.fail(TIMEOUT);
+ }
+ return result;
+ }
+
+ @Override
+ public CompletableFuture<MeterStoreResult> updateMeter(Meter meter) {
+ CompletableFuture<MeterStoreResult> future = new CompletableFuture<>();
+ MeterKey key = MeterKey.key(meter.deviceId(), meter.id());
+ futures.put(key, future);
+
+ MeterData data = new MeterData(meter, null, local);
+ try {
+ if (meters.computeIfPresent(key, (k, v) -> data) == null) {
+ future.complete(MeterStoreResult.fail(MeterFailReason.INVALID_METER));
+ }
+ } catch (StorageException e) {
+ futures.remove(key);
+ future.completeExceptionally(e);
+ }
+ return future;
+ }
+
+ @Override
+ public void updateMeterState(Meter meter) {
+ MeterKey key = MeterKey.key(meter.deviceId(), meter.id());
+ meters.computeIfPresent(key, (k, v) -> {
+ DefaultMeter m = (DefaultMeter) v.meter();
+ m.setState(meter.state());
+ m.setProcessedPackets(meter.packetsSeen());
+ m.setProcessedBytes(meter.bytesSeen());
+ m.setLife(meter.life());
+ // TODO: Prune if drops to zero.
+ m.setReferenceCount(meter.referenceCount());
+ return new MeterData(m, null, v.origin());
+ });
+ }
+
+ @Override
+ public Meter getMeter(MeterKey key) {
+ MeterData data = Versioned.valueOrElse(meters.get(key), null);
+ return data == null ? null : data.meter();
+ }
+
+ @Override
+ public Collection<Meter> getAllMeters() {
+ return Collections2.transform(meters.asJavaMap().values(),
+ MeterData::meter);
+ }
+
+ @Override
+ public Collection<Meter> getAllMeters(DeviceId deviceId) {
+ return Collections2.transform(
+ Collections2.filter(meters.asJavaMap().values(),
+ (MeterData m) -> m.meter().deviceId().equals(deviceId)),
+ MeterData::meter);
+ }
+
+ @Override
+ public void failedMeter(MeterOperation op, MeterFailReason reason) {
+ MeterKey key = MeterKey.key(op.meter().deviceId(), op.meter().id());
+ meters.computeIfPresent(key, (k, v) ->
+ new MeterData(v.meter(), reason, v.origin()));
+ }
+
+ @Override
+ public void deleteMeterNow(Meter m) {
+ // Create the key
+ MeterKey key = MeterKey.key(m.deviceId(), m.id());
+ // Remove the future
+ futures.remove(key);
+ // Remove the meter
+ meters.remove(key);
+ // Free the id
+ freeMeterId(m.deviceId(), m.id());
+ // Finally notify the delegate
+ notifyDelegate(new MeterEvent(MeterEvent.Type.METER_REMOVED, m));
+ }
+
+ @Override
+ public long getMaxMeters(MeterFeaturesKey key) {
+ MeterFeatures features = Versioned.valueOrElse(meterFeatures.get(key), null);
+ return features == null ? 0L : features.maxMeter();
+ }
+
+ // queryMaxMeters is implemented in FullMetersAvailable behaviour.
+ private long queryMaxMeters(DeviceId device) {
+ // Get driver handler for this device
+ DriverHandler handler = driverService.createHandler(device);
+ // If creation failed or the device does not have this behavior
+ if (handler == null || !handler.hasBehaviour(MeterQuery.class)) {
+ // We cannot know max meter
+ return 0L;
+ }
+ // Get the behavior
+ MeterQuery query = handler.behaviour(MeterQuery.class);
+ // Return as max meter the result of the query
+ return query.getMaxMeters();
+ }
+
+ private boolean updateMeterIdAvailability(DeviceId deviceId, MeterId id,
+ boolean available) {
+ // According to available, make available or unavailable a meter key
+ return available ? availableMeterIds.add(MeterKey.key(deviceId, id)) :
+ availableMeterIds.remove(MeterKey.key(deviceId, id));
+ }
+
+ private MeterId getNextAvailableId(Set<MeterId> availableIds) {
+ // If there are no available ids
+ if (availableIds.isEmpty()) {
+ // Just end the cycle
+ return null;
+ }
+ // If it is the first fit
+ if (reuseStrategy == FIRST_FIT || availableIds.size() == 1) {
+ return availableIds.iterator().next();
+ }
+ // If it is random, get the size
+ int size = availableIds.size();
+ // Return a random element
+ return Iterables.get(availableIds, RandomUtils.nextInt(size));
+ }
+
+ // Implements reuse strategy
+ private MeterId firstReusableMeterId(DeviceId deviceId) {
+ // Filter key related to device id, and reduce to meter ids
+ Set<MeterId> localAvailableMeterIds = availableMeterIds.stream()
+ .filter(meterKey -> meterKey.deviceId().equals(deviceId))
+ .map(MeterKey::meterId)
+ .collect(Collectors.toSet());
+ // Get next available id
+ MeterId meterId = getNextAvailableId(localAvailableMeterIds);
+ // Iterate until there are items
+ while (meterId != null) {
+ // If we are able to reserve the id
+ if (updateMeterIdAvailability(deviceId, meterId, false)) {
+ // Just end
+ return meterId;
+ }
+ // Update the set
+ localAvailableMeterIds.remove(meterId);
+ // Try another time
+ meterId = getNextAvailableId(localAvailableMeterIds);
+ }
+ // No reusable ids
+ return null;
+ }
+
+ @Override
+ public MeterId allocateMeterId(DeviceId deviceId) {
+ // Init steps
+ MeterId meterId;
+ long id;
+ // Try to reuse meter id
+ meterId = firstReusableMeterId(deviceId);
+ // We found a reusable id, return
+ if (meterId != null) {
+ return meterId;
+ }
+ // If there was no reusable MeterId we have to generate a new value
+ // using maxMeters as upper limit.
+ long maxMeters = getMaxMeters(MeterFeaturesKey.key(deviceId));
+ // If the device does not give us MeterFeatures
+ if (maxMeters == 0L) {
+ // MeterFeatures couldn't be retrieved, fallback to queryMeters.
+ maxMeters = queryMaxMeters(deviceId);
+ }
+ // If we don't know the max, cannot proceed
+ if (maxMeters == 0L) {
+ return null;
+ }
+ // Get a new value
+ id = meterIdGenerators.incrementAndGet(deviceId);
+ // Check with the max, and if the value is bigger, cannot proceed
+ if (id >= maxMeters) {
+ return null;
+ }
+ // Done, return the value
+ return MeterId.meterId(id);
+ }
+
+ @Override
+ public void freeMeterId(DeviceId deviceId, MeterId meterId) {
+ // Avoid to free meter not allocated
+ if (meterIdGenerators.get(deviceId) < meterId.id()) {
+ return;
+ }
+ // Update the availability
+ updateMeterIdAvailability(deviceId, meterId, true);
+ }
+
+ private class InternalMapEventListener implements MapEventListener<MeterKey, MeterData> {
+ @Override
+ public void event(MapEvent<MeterKey, MeterData> event) {
+ MeterKey key = event.key();
+ Versioned<MeterData> value = event.type() == MapEvent.Type.REMOVE ? event.oldValue() : event.newValue();
+ MeterData data = value.value();
+ NodeId master = mastershipService.getMasterFor(data.meter().deviceId());
+ switch (event.type()) {
+ case INSERT:
+ case UPDATE:
+ switch (data.meter().state()) {
+ case PENDING_ADD:
+ case PENDING_REMOVE:
+ if (!data.reason().isPresent() && local.equals(master)) {
+ notifyDelegate(
+ new MeterEvent(data.meter().state() == MeterState.PENDING_ADD ?
+ MeterEvent.Type.METER_ADD_REQ : MeterEvent.Type.METER_REM_REQ,
+ data.meter()));
+ } else if (data.reason().isPresent() && local.equals(data.origin())) {
+ MeterStoreResult msr = MeterStoreResult.fail(data.reason().get());
+ //TODO: No future -> no friend
+ futures.get(key).complete(msr);
+ }
+ break;
+ case ADDED:
+ if (local.equals(data.origin()) &&
+ (data.meter().state() == MeterState.PENDING_ADD
+ || data.meter().state() == MeterState.ADDED)) {
+ futures.computeIfPresent(key, (k, v) -> {
+ notifyDelegate(
+ new MeterEvent(MeterEvent.Type.METER_ADDED, data.meter()));
+ return null;
+ });
+ }
+ break;
+ case REMOVED:
+ if (local.equals(data.origin()) && data.meter().state() == MeterState.PENDING_REMOVE) {
+ futures.remove(key).complete(MeterStoreResult.success());
+ }
+ break;
+ default:
+ log.warn("Unknown meter state type {}", data.meter().state());
+ }
+ break;
+ case REMOVE:
+ //Only happens at origin so we do not need to care.
+ break;
+ default:
+ log.warn("Unknown Map event type {}", event.type());
+ }
+
+ }
+ }
+
+
+}
diff --git a/core/store/dist/src/main/java/org/onosproject/store/meter/impl/MeterData.java b/core/store/dist/src/main/java/org/onosproject/store/meter/impl/MeterData.java
new file mode 100644
index 0000000..8f87c2c
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onosproject/store/meter/impl/MeterData.java
@@ -0,0 +1,52 @@
+/*
+ * Copyright 2015-present Open Networking Foundation
+ *
+ * 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.meter.impl;
+
+import org.onosproject.cluster.NodeId;
+import org.onosproject.net.meter.Meter;
+import org.onosproject.net.meter.MeterFailReason;
+
+import java.util.Optional;
+
+/**
+ * A class representing the meter information stored in the meter store.
+ */
+public class MeterData {
+
+ private final Meter meter;
+ private final Optional<MeterFailReason> reason;
+ private final NodeId origin;
+
+ public MeterData(Meter meter, MeterFailReason reason, NodeId origin) {
+ this.meter = meter;
+ this.reason = Optional.ofNullable(reason);
+ this.origin = origin;
+ }
+
+ public Meter meter() {
+ return meter;
+ }
+
+ public Optional<MeterFailReason> reason() {
+ return this.reason;
+ }
+
+ public NodeId origin() {
+ return this.origin;
+ }
+
+
+}
diff --git a/core/store/dist/src/main/java/org/onosproject/store/meter/impl/package-info.java b/core/store/dist/src/main/java/org/onosproject/store/meter/impl/package-info.java
new file mode 100644
index 0000000..bfd490b
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onosproject/store/meter/impl/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * Copyright 2015-present Open Networking Foundation
+ *
+ * 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.
+ */
+
+/**
+ * A distributed meter store implementation that stores meter data consistently
+ * across the cluster.
+ */
+package org.onosproject.store.meter.impl;
diff --git a/core/store/dist/src/test/java/org/onosproject/store/meter/impl/DistributedMeterStoreTest.java b/core/store/dist/src/test/java/org/onosproject/store/meter/impl/DistributedMeterStoreTest.java
new file mode 100644
index 0000000..5005787
--- /dev/null
+++ b/core/store/dist/src/test/java/org/onosproject/store/meter/impl/DistributedMeterStoreTest.java
@@ -0,0 +1,502 @@
+/*
+ * Copyright 2017-present Open Networking Foundation
+ *
+ * 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.meter.impl;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.onlab.junit.TestUtils;
+import org.onlab.packet.IpAddress;
+import org.onlab.util.KryoNamespace;
+import org.onosproject.TestApplicationId;
+import org.onosproject.cluster.ClusterServiceAdapter;
+import org.onosproject.cluster.ControllerNode;
+import org.onosproject.cluster.DefaultControllerNode;
+import org.onosproject.cluster.NodeId;
+import org.onosproject.mastership.MastershipServiceAdapter;
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.behaviour.MeterQuery;
+import org.onosproject.net.driver.Behaviour;
+import org.onosproject.net.driver.Driver;
+import org.onosproject.net.driver.DriverData;
+import org.onosproject.net.driver.DriverHandler;
+import org.onosproject.net.driver.DriverServiceAdapter;
+import org.onosproject.net.meter.Band;
+import org.onosproject.net.meter.DefaultBand;
+import org.onosproject.net.meter.DefaultMeter;
+import org.onosproject.net.meter.DefaultMeterFeatures;
+import org.onosproject.net.meter.Meter;
+import org.onosproject.net.meter.MeterFeatures;
+import org.onosproject.net.meter.MeterFeaturesKey;
+import org.onosproject.net.meter.MeterId;
+import org.onosproject.net.meter.MeterKey;
+import org.onosproject.net.meter.MeterState;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.TestStorageService;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.*;
+import static org.onosproject.net.NetTestTools.APP_ID;
+import static org.onosproject.net.NetTestTools.did;
+
+/**
+ * Meter store tests.
+ */
+public class DistributedMeterStoreTest {
+
+ // Test node id
+ private static final NodeId NID_LOCAL = new NodeId("local");
+
+ // Test ip address
+ private static final IpAddress LOCALHOST = IpAddress.valueOf("127.0.0.1");
+
+ // Store under testing
+ private DistributedMeterStore meterStore;
+
+ // Device ids used during the tests
+ private DeviceId did1 = did("1");
+ private DeviceId did2 = did("2");
+ private DeviceId did3 = did("3");
+ private DeviceId did4 = did("4");
+
+ // Meter ids used during the tests
+ private MeterId mid1 = MeterId.meterId(1);
+ private MeterId mid2 = MeterId.meterId(2);
+ private MeterId mid10 = MeterId.meterId(10);
+
+ // Bands used during the tests
+ private Band b1 = DefaultBand.builder()
+ .ofType(Band.Type.DROP)
+ .withRate(500)
+ .build();
+
+ // Meters used during the tests
+ private Meter m1 = DefaultMeter.builder()
+ .forDevice(did1)
+ .fromApp(APP_ID)
+ .withId(mid1)
+ .withUnit(Meter.Unit.KB_PER_SEC)
+ .withBands(Collections.singletonList(b1))
+ .build();
+
+ // Meter features used during the tests
+ private MeterFeatures mef1 = DefaultMeterFeatures.builder().forDevice(did1)
+ .withMaxMeters(3L)
+ .withBandTypes(new HashSet<>())
+ .withUnits(new HashSet<>())
+ .hasStats(false)
+ .hasBurst(false)
+ .withMaxBands((byte) 0)
+ .withMaxColors((byte) 0)
+ .build();
+ private MeterFeatures mef2 = DefaultMeterFeatures.builder().forDevice(did2)
+ .withMaxMeters(10L)
+ .withBandTypes(new HashSet<>())
+ .withUnits(new HashSet<>())
+ .hasStats(false)
+ .hasBurst(false)
+ .withMaxBands((byte) 0)
+ .withMaxColors((byte) 0)
+ .build();
+
+ @Before
+ public void setup() {
+ // Init step
+ meterStore = new DistributedMeterStore();
+ // Let's initialize some internal services
+ TestUtils.setField(meterStore, "storageService", new TestStorageService());
+ TestUtils.setField(meterStore, "clusterService", new TestClusterService());
+ TestUtils.setField(meterStore, "mastershipService", new TestMastershipService());
+ TestUtils.setField(meterStore, "driverService", new TestDriverService());
+
+ // Inject TestApplicationId into the DistributedMeterStore serializer
+ KryoNamespace.Builder testKryoBuilder = TestUtils.getField(meterStore, "APP_KRYO_BUILDER");
+ testKryoBuilder.register(TestApplicationId.class);
+ Serializer testSerializer = Serializer.using(Lists.newArrayList(testKryoBuilder.build()));
+ TestUtils.setField(meterStore, "serializer", testSerializer);
+
+ // Activate the store
+ meterStore.activate();
+ }
+
+ @After
+ public void tearDown() {
+ // Deactivate the store
+ meterStore.deactivate();
+ }
+
+ private void initMeterStore() {
+ // Let's store feature for device 1
+ meterStore.storeMeterFeatures(mef1);
+ // Let's store feature for device 2
+ meterStore.storeMeterFeatures(mef2);
+ }
+
+ /**
+ * Test proper store of meter features.
+ */
+ @Test
+ public void testStoreMeterFeatures() {
+ // Let's store feature for device 1
+ meterStore.storeMeterFeatures(mef1);
+ // Verify store meter features
+ assertThat(meterStore.getMaxMeters(MeterFeaturesKey.key(did1)), is(3L));
+ // Let's store feature for device 1
+ meterStore.storeMeterFeatures(mef2);
+ // Verify store meter features
+ assertThat(meterStore.getMaxMeters(MeterFeaturesKey.key(did2)), is(10L));
+ }
+
+ /**
+ * Test proper delete of meter features.
+ */
+ @Test
+ public void testDeleteMeterFeatures() {
+ // Let's store feature for device 1
+ meterStore.storeMeterFeatures(mef1);
+ // Verify store meter features
+ assertThat(meterStore.getMaxMeters(MeterFeaturesKey.key(did1)), is(3L));
+ // Let's delete the features
+ meterStore.deleteMeterFeatures(did1);
+ // Verify delete meter features
+ assertThat(meterStore.getMaxMeters(MeterFeaturesKey.key(did1)), is(0L));
+ }
+
+ /**
+ * Test proper allocation of meter ids.
+ */
+ @Test
+ public void testAllocateId() {
+ // Init the store
+ initMeterStore();
+ // Allocate a meter id and verify is equal to mid1
+ assertThat(mid1, is(meterStore.allocateMeterId(did1)));
+ // Allocate a meter id and verify is equal to mid2
+ assertThat(mid2, is(meterStore.allocateMeterId(did1)));
+ }
+
+ /**
+ * Test proper free of meter ids.
+ */
+ @Test
+ public void testFreeId() {
+ // Init the store
+ initMeterStore();
+ // Allocate a meter id and verify is equal to mid1
+ assertThat(mid1, is(meterStore.allocateMeterId(did1)));
+ // Free the above id
+ meterStore.freeMeterId(did1, mid1);
+ // Allocate a meter id and verify is equal to mid1
+ assertThat(mid1, is(meterStore.allocateMeterId(did1)));
+ // Free an id not allocated
+ meterStore.freeMeterId(did1, mid10);
+ // Allocate a meter id and verify is equal to mid2
+ assertThat(mid2, is(meterStore.allocateMeterId(did1)));
+ }
+
+ /**
+ * Test proper reuse of meter ids.
+ */
+ @Test
+ public void testReuseId() {
+ // Init the store
+ initMeterStore();
+ // Reserve id 1
+ MeterId meterIdOne = meterStore.allocateMeterId(did2);
+ // Free the above id
+ meterStore.freeMeterId(did2, meterIdOne);
+ // Start an async reservation
+ CompletableFuture<MeterId> future = CompletableFuture.supplyAsync(
+ () -> meterStore.allocateMeterId(did2)
+ );
+ // Start another reservation
+ MeterId meterIdTwo = meterStore.allocateMeterId(did2);
+ try {
+ meterIdOne = future.get();
+ } catch (InterruptedException | ExecutionException e) {
+ e.printStackTrace();
+ }
+ // Ids should be different, otherwise we had clash in the store
+ assertNotEquals("Ids should be different", meterIdOne, meterIdTwo);
+
+ // Free the above id
+ meterStore.freeMeterId(did1, meterIdOne);
+ // Free the above id
+ meterStore.freeMeterId(did1, meterIdTwo);
+ // Reserve id 1
+ meterIdOne = meterStore.allocateMeterId(did2);
+ // Reserve id 2
+ meterStore.allocateMeterId(did2);
+ // Reserve id 3
+ MeterId meterIdThree = meterStore.allocateMeterId(did2);
+ // Reserve id 4
+ MeterId meterIdFour = meterStore.allocateMeterId(did2);
+ // Free the above id
+ meterStore.freeMeterId(did1, meterIdOne);
+ // Free the above id
+ meterStore.freeMeterId(did1, meterIdThree);
+ // Free the above id
+ meterStore.freeMeterId(did1, meterIdFour);
+ // Start an async reservation
+ future = CompletableFuture.supplyAsync(
+ () -> meterStore.allocateMeterId(did2)
+ );
+ // Start another reservation
+ MeterId meterAnotherId = meterStore.allocateMeterId(did2);
+ try {
+ meterAnotherId = future.get();
+ } catch (InterruptedException | ExecutionException e) {
+ e.printStackTrace();
+ }
+ // Ids should be different, otherwise we had clash in the store
+ assertNotEquals("Ids should be different", meterAnotherId, meterIdOne);
+ }
+
+ /**
+ * Test query meters mechanism.
+ */
+ @Test
+ public void testQueryMeters() {
+ // Init the store
+ initMeterStore();
+ // Let's test queryMeters
+ assertThat(mid1, is(meterStore.allocateMeterId(did3)));
+ // Let's test queryMeters error
+ assertNull(meterStore.allocateMeterId(did4));
+ }
+
+ /**
+ * Test max meter error.
+ */
+ @Test
+ public void testMaxMeterError() {
+ // Init the store
+ initMeterStore();
+ // Reserve id 1
+ assertThat(mid1, is(meterStore.allocateMeterId(did1)));
+ // Reserve id 2
+ assertThat(mid2, is(meterStore.allocateMeterId(did1)));
+ // Max meter error
+ assertNull(meterStore.allocateMeterId(did1));
+ }
+
+ /**
+ * Test store meter.
+ */
+ @Test
+ public void testStoreMeter() {
+ // Init the store
+ initMeterStore();
+ // Simulate the allocation of an id
+ MeterId idOne = meterStore.allocateMeterId(did1);
+ // Verify the allocation
+ assertThat(mid1, is(idOne));
+ // Let's create a meter
+ Meter meterOne = DefaultMeter.builder()
+ .forDevice(did1)
+ .fromApp(APP_ID)
+ .withId(mid1)
+ .withUnit(Meter.Unit.KB_PER_SEC)
+ .withBands(Collections.singletonList(b1))
+ .build();
+ // Set the state
+ ((DefaultMeter) meterOne).setState(MeterState.PENDING_ADD);
+ // Store the meter
+ meterStore.storeMeter(meterOne);
+ // Let's create meter key
+ MeterKey meterKey = MeterKey.key(did1, mid1);
+ // Verify the store
+ assertThat(1, is(meterStore.getAllMeters().size()));
+ assertThat(1, is(meterStore.getAllMeters(did1).size()));
+ assertThat(m1, is(meterStore.getMeter(meterKey)));
+ }
+
+ /**
+ * Test delete meter.
+ */
+ @Test
+ public void testDeleteMeter() {
+ // Init the store
+ initMeterStore();
+ // Simulate the allocation of an id
+ MeterId idOne = meterStore.allocateMeterId(did1);
+ // Verify the allocation
+ assertThat(mid1, is(idOne));
+ // Let's create a meter
+ Meter meterOne = DefaultMeter.builder()
+ .forDevice(did1)
+ .fromApp(APP_ID)
+ .withId(mid1)
+ .withUnit(Meter.Unit.KB_PER_SEC)
+ .withBands(Collections.singletonList(b1))
+ .build();
+ // Set the state
+ ((DefaultMeter) meterOne).setState(MeterState.PENDING_ADD);
+ // Store the meter
+ meterStore.storeMeter(meterOne);
+ // Set the state
+ ((DefaultMeter) meterOne).setState(MeterState.PENDING_REMOVE);
+ // Let's create meter key
+ MeterKey meterKey = MeterKey.key(did1, mid1);
+ // Delete meter
+ meterStore.deleteMeter(meterOne);
+ // Start an async delete, simulating the operation of the provider
+ CompletableFuture<Void> future = CompletableFuture.runAsync(
+ () -> meterStore.deleteMeterNow(meterOne)
+ );
+ // Let's wait
+ try {
+ future.get();
+ } catch (InterruptedException | ExecutionException e) {
+ e.printStackTrace();
+ }
+ // Verify delete
+ assertThat(0, is(meterStore.getAllMeters().size()));
+ assertThat(0, is(meterStore.getAllMeters(did1).size()));
+ assertNull(meterStore.getMeter(meterKey));
+ assertThat(mid1, is(meterStore.allocateMeterId(did1)));
+ }
+
+ /**
+ * Test no delete meter.
+ */
+ @Test
+ public void testNoDeleteMeter() {
+ // Init the store
+ initMeterStore();
+ // Simulate the allocation of an id
+ MeterId idOne = meterStore.allocateMeterId(did1);
+ // Create the key
+ MeterKey keyOne = MeterKey.key(did1, idOne);
+ // Let's create a meter
+ Meter meterOne = DefaultMeter.builder()
+ .forDevice(did1)
+ .fromApp(APP_ID)
+ .withId(mid1)
+ .withUnit(Meter.Unit.KB_PER_SEC)
+ .withBands(Collections.singletonList(b1))
+ .build();
+ // Set the state
+ ((DefaultMeter) meterOne).setState(MeterState.PENDING_REMOVE);
+ // Delete meter
+ meterStore.deleteMeter(meterOne);
+ // Verify No delete
+ assertThat(0, is(meterStore.getAllMeters().size()));
+ assertThat(0, is(meterStore.getAllMeters(did1).size()));
+ assertNull(meterStore.getMeter(keyOne));
+ }
+
+ // Test cluster service
+ private final class TestClusterService extends ClusterServiceAdapter {
+
+ private ControllerNode local = new DefaultControllerNode(NID_LOCAL, LOCALHOST);
+
+ @Override
+ public ControllerNode getLocalNode() {
+ return local;
+ }
+
+ @Override
+ public Set<ControllerNode> getNodes() {
+ return Sets.newHashSet();
+ }
+
+ }
+
+ // Test mastership service
+ private final class TestMastershipService extends MastershipServiceAdapter {
+ @Override
+ public NodeId getMasterFor(DeviceId deviceId) {
+ return NID_LOCAL;
+ }
+ }
+
+ // Test class for driver service.
+ private class TestDriverService extends DriverServiceAdapter {
+ @Override
+ public DriverHandler createHandler(DeviceId deviceId, String... credentials) {
+ return deviceId.equals(did3) ? new TestDriverHandler() : null;
+ }
+ }
+
+ // Test class for driver handler.
+ private class TestDriverHandler implements DriverHandler {
+
+ @Override
+ public Driver driver() {
+ return null;
+ }
+
+ @Override
+ public DriverData data() {
+ return null;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T extends Behaviour> T behaviour(Class<T> behaviourClass) {
+ return (T) new TestMeterQuery();
+ }
+
+ @Override
+ public <T> T get(Class<T> serviceClass) {
+ return null;
+ }
+
+ @Override
+ public boolean hasBehaviour(Class<? extends Behaviour> behaviourClass) {
+ return true;
+ }
+ }
+
+ // Test meter query
+ private class TestMeterQuery implements MeterQuery {
+
+ @Override
+ public DriverData data() {
+ return null;
+ }
+
+ @Override
+ public void setData(DriverData data) {
+
+ }
+ @Override
+ public DriverHandler handler() {
+ return null;
+ }
+
+ @Override
+ public void setHandler(DriverHandler handler) {
+
+ }
+
+ @Override
+ public long getMaxMeters() {
+ return 100;
+ }
+ }
+
+}