ECDeviceStore: DeviceStore built using ONOS distributed primitives: ECMap and Di…

…stributedSet (disabled right now)

Change-Id: I36fdcd635f982f2b8dac291c52be4662601ef9f0

+/*
+ * Copyright 2014-2015 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.device.impl;
+
+import java.util.Objects;
+
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.provider.ProviderId;
+
+import com.google.common.base.MoreObjects;
+
+/**
+ * Key for DeviceDescriptions in ECDeviceStore.
+ */
+public class DeviceKey {
+    private final ProviderId providerId;
+    private final DeviceId deviceId;
+
+    public DeviceKey(ProviderId providerId, DeviceId deviceId) {
+        this.providerId = providerId;
+        this.deviceId = deviceId;
+    }
+
+    public ProviderId providerId() {
+        return providerId;
+    }
+
+    public DeviceId deviceId() {
+        return deviceId;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(providerId, deviceId);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (!(obj instanceof DeviceKey)) {
+            return false;
+        }
+        DeviceKey that = (DeviceKey) obj;
+        return Objects.equals(this.deviceId, that.deviceId) &&
+               Objects.equals(this.providerId, that.providerId);
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(getClass())
+                .add("providerId", providerId)
+                .add("deviceId", deviceId)
+                .toString();
+    }
+}

+/*
+ * Copyright 2014-2015 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.device.impl;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Verify.verify;
+import static org.onosproject.net.DefaultAnnotations.merge;
+import static org.slf4j.LoggerFactory.getLogger;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.Map.Entry;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.apache.felix.scr.annotations.Service;
+import org.onlab.packet.ChassisId;
+import org.onlab.util.KryoNamespace;
+import org.onlab.util.SharedExecutors;
+import org.onosproject.cluster.ClusterService;
+import org.onosproject.cluster.NodeId;
+import org.onosproject.mastership.MastershipService;
+import org.onosproject.mastership.MastershipTermService;
+import org.onosproject.net.Annotations;
+import org.onosproject.net.AnnotationsUtil;
+import org.onosproject.net.DefaultAnnotations;
+import org.onosproject.net.DefaultDevice;
+import org.onosproject.net.DefaultPort;
+import org.onosproject.net.Device;
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.MastershipRole;
+import org.onosproject.net.OchPort;
+import org.onosproject.net.OduCltPort;
+import org.onosproject.net.OmsPort;
+import org.onosproject.net.Port;
+import org.onosproject.net.PortNumber;
+import org.onosproject.net.Device.Type;
+import org.onosproject.net.device.DefaultPortStatistics;
+import org.onosproject.net.device.DeviceClockService;
+import org.onosproject.net.device.DeviceDescription;
+import org.onosproject.net.device.DeviceEvent;
+import org.onosproject.net.device.DeviceStore;
+import org.onosproject.net.device.DeviceStoreDelegate;
+import org.onosproject.net.device.OchPortDescription;
+import org.onosproject.net.device.OduCltPortDescription;
+import org.onosproject.net.device.OmsPortDescription;
+import org.onosproject.net.device.PortDescription;
+import org.onosproject.net.device.PortStatistics;
+import org.onosproject.net.provider.ProviderId;
+import org.onosproject.store.AbstractStore;
+import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
+import org.onosproject.store.impl.MastershipBasedTimestamp;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.serializers.KryoSerializer;
+import org.onosproject.store.serializers.custom.DistributedStoreSerializers;
+import org.onosproject.store.service.DistributedSet;
+import org.onosproject.store.service.EventuallyConsistentMap;
+import org.onosproject.store.service.EventuallyConsistentMapEvent;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.SetEvent;
+import org.onosproject.store.service.SetEventListener;
+import org.onosproject.store.service.WallClockTimestamp;
+
+import static org.onosproject.store.service.EventuallyConsistentMapEvent.Type.PUT;
+import static org.onosproject.store.service.EventuallyConsistentMapEvent.Type.REMOVE;
+
+import org.onosproject.store.service.EventuallyConsistentMapListener;
+import org.onosproject.store.service.StorageService;
+import org.slf4j.Logger;
+
+import static org.onosproject.net.device.DeviceEvent.Type.*;
+import static org.onosproject.store.device.impl.GossipDeviceStoreMessageSubjects.DEVICE_INJECTED;
+import static org.onosproject.store.device.impl.GossipDeviceStoreMessageSubjects.DEVICE_REMOVE_REQ;
+import static org.onosproject.store.device.impl.GossipDeviceStoreMessageSubjects.PORT_INJECTED;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.Futures;
+
+/**
+ * Manages the inventory of devices using a {@code EventuallyConsistentMap}.
+ */
+@Component(immediate = true, enabled = false)
+@Service
+public class ECDeviceStore
+    extends AbstractStore<DeviceEvent, DeviceStoreDelegate>
+    implements DeviceStore {
+
+    private final Logger log = getLogger(getClass());
+
+    private static final String DEVICE_NOT_FOUND = "Device with ID %s not found";
+
+    private final Map<DeviceId, Device> devices = Maps.newConcurrentMap();
+    private final Map<DeviceId, Map<PortNumber, Port>> devicePorts = Maps.newConcurrentMap();
+    Set<DeviceId> pendingAvailableChangeUpdates = Sets.newConcurrentHashSet();
+
+    private EventuallyConsistentMap<DeviceKey, DeviceDescription> deviceDescriptions;
+    private EventuallyConsistentMap<PortKey, PortDescription> portDescriptions;
+    private EventuallyConsistentMap<DeviceId, Map<PortNumber, PortStatistics>> devicePortStats;
+    private EventuallyConsistentMap<DeviceId, Map<PortNumber, PortStatistics>> devicePortDeltaStats;
+
+    private DistributedSet<DeviceId> availableDevices;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected StorageService storageService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected MastershipService mastershipService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected MastershipTermService mastershipTermService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected DeviceClockService deviceClockService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected ClusterCommunicationService clusterCommunicator;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected ClusterService clusterService;
+
+    private NodeId localNodeId;
+    private EventuallyConsistentMapListener<DeviceKey, DeviceDescription> deviceUpdateListener =
+            new InternalDeviceChangeEventListener();
+    private EventuallyConsistentMapListener<PortKey, PortDescription> portUpdateListener =
+            new InternalPortChangeEventListener();
+    private final EventuallyConsistentMapListener<DeviceId, Map<PortNumber, PortStatistics>> portStatsListener =
+            new InternalPortStatsListener();
+    private final SetEventListener<DeviceId> deviceStatusTracker =
+            new InternalDeviceStatusTracker();
+
+    protected static final KryoSerializer SERIALIZER = new KryoSerializer() {
+        @Override
+        protected void setupKryoPool() {
+            serializerPool = KryoNamespace.newBuilder()
+                    .register(DistributedStoreSerializers.STORE_COMMON)
+                    .nextId(DistributedStoreSerializers.STORE_CUSTOM_BEGIN)
+                    .register(DeviceInjectedEvent.class)
+                    .register(PortInjectedEvent.class)
+                    .build();
+        }
+    };
+
+    protected static final KryoNamespace.Builder SERIALIZER_BUILDER = KryoNamespace.newBuilder()
+            .register(KryoNamespaces.API)
+            .register(DeviceKey.class)
+            .register(PortKey.class)
+            .register(DeviceKey.class)
+            .register(PortKey.class)
+            .register(MastershipBasedTimestamp.class);
+
+    @Activate
+    public void activate() {
+        localNodeId = clusterService.getLocalNode().id();
+
+        deviceDescriptions = storageService.<DeviceKey, DeviceDescription>eventuallyConsistentMapBuilder()
+                .withName("onos-device-descriptions")
+                .withSerializer(SERIALIZER_BUILDER)
+                .withTimestampProvider((k, v) -> {
+                    try {
+                        return deviceClockService.getTimestamp(k.deviceId());
+                    } catch (IllegalStateException e) {
+                        return null;
+                    }
+                }).build();
+
+        portDescriptions = storageService.<PortKey, PortDescription>eventuallyConsistentMapBuilder()
+                .withName("onos-port-descriptions")
+                .withSerializer(SERIALIZER_BUILDER)
+                .withTimestampProvider((k, v) -> {
+                    try {
+                        return deviceClockService.getTimestamp(k.deviceId());
+                    } catch (IllegalStateException e) {
+                        return null;
+                    }
+                }).build();
+
+        devicePortStats = storageService.<DeviceId, Map<PortNumber, PortStatistics>>eventuallyConsistentMapBuilder()
+                .withName("onos-port-stats")
+                .withSerializer(SERIALIZER_BUILDER)
+                .withAntiEntropyPeriod(5, TimeUnit.SECONDS)
+                .withTimestampProvider((k, v) -> new WallClockTimestamp())
+                .withTombstonesDisabled()
+                .build();
+
+        devicePortDeltaStats = storageService.<DeviceId, Map<PortNumber, PortStatistics>>
+                eventuallyConsistentMapBuilder()
+                .withName("onos-port-stats-delta")
+                .withSerializer(SERIALIZER_BUILDER)
+                .withAntiEntropyPeriod(5, TimeUnit.SECONDS)
+                .withTimestampProvider((k, v) -> new WallClockTimestamp())
+                .withTombstonesDisabled()
+                .build();
+
+        clusterCommunicator.addSubscriber(DEVICE_INJECTED,
+                SERIALIZER::decode,
+                this::injectDevice,
+                SERIALIZER::encode,
+                SharedExecutors.getPoolThreadExecutor());
+
+        clusterCommunicator.addSubscriber(PORT_INJECTED,
+                SERIALIZER::decode,
+                this::injectPort,
+                SERIALIZER::encode,
+                SharedExecutors.getPoolThreadExecutor());
+
+        availableDevices = storageService.<DeviceId>setBuilder()
+                .withName("onos-online-devices")
+                .withSerializer(Serializer.using(KryoNamespaces.API))
+                .withPartitionsDisabled()
+                .withRelaxedReadConsistency()
+                .build();
+
+        deviceDescriptions.addListener(deviceUpdateListener);
+        portDescriptions.addListener(portUpdateListener);
+        devicePortStats.addListener(portStatsListener);
+        availableDevices.addListener(deviceStatusTracker);
+        log.info("Started");
+    }
+
+    @Deactivate
+    public void deactivate() {
+        devicePortStats.removeListener(portStatsListener);
+        deviceDescriptions.removeListener(deviceUpdateListener);
+        portDescriptions.removeListener(portUpdateListener);
+        availableDevices.removeListener(deviceStatusTracker);
+        devicePortStats.destroy();
+        devicePortDeltaStats.destroy();
+        deviceDescriptions.destroy();
+        portDescriptions.destroy();
+        devices.clear();
+        devicePorts.clear();
+        clusterCommunicator.removeSubscriber(DEVICE_INJECTED);
+        clusterCommunicator.removeSubscriber(PORT_INJECTED);
+        log.info("Stopped");
+    }
+
+    @Override
+    public Iterable<Device> getDevices() {
+        return devices.values();
+    }
+
+    @Override
+    public int getDeviceCount() {
+        return devices.size();
+    }
+
+    @Override
+    public Device getDevice(DeviceId deviceId) {
+        return devices.get(deviceId);
+    }
+
+    @Override
+    public DeviceEvent createOrUpdateDevice(ProviderId providerId,
+            DeviceId deviceId,
+            DeviceDescription deviceDescription) {
+        NodeId master = mastershipService.getMasterFor(deviceId);
+        if (localNodeId.equals(master)) {
+            deviceDescriptions.put(new DeviceKey(providerId, deviceId), deviceDescription);
+            return refreshDeviceCache(providerId, deviceId);
+        } else {
+            DeviceInjectedEvent deviceInjectedEvent = new DeviceInjectedEvent(providerId, deviceId, deviceDescription);
+            return Futures.getUnchecked(
+                    clusterCommunicator.sendAndReceive(deviceInjectedEvent,
+                            DEVICE_INJECTED,
+                            SERIALIZER::encode,
+                            SERIALIZER::decode,
+                            master));
+        }
+    }
+
+    private DeviceEvent refreshDeviceCache(ProviderId providerId, DeviceId deviceId) {
+        AtomicReference<DeviceEvent.Type> eventType = new AtomicReference<>();
+        Device device = devices.compute(deviceId, (k, existingDevice) -> {
+            Device newDevice = composeDevice(deviceId);
+            if (existingDevice == null) {
+                eventType.set(DEVICE_ADDED);
+            } else {
+                // We allow only certain attributes to trigger update
+                boolean propertiesChanged =
+                        !Objects.equals(existingDevice.hwVersion(), newDevice.hwVersion()) ||
+                                !Objects.equals(existingDevice.swVersion(), newDevice.swVersion()) ||
+                                !Objects.equals(existingDevice.providerId(), newDevice.providerId());
+                boolean annotationsChanged =
+                        !AnnotationsUtil.isEqual(existingDevice.annotations(), newDevice.annotations());
+
+                // Primary providers can respond to all changes, but ancillary ones
+                // should respond only to annotation changes.
+                if ((providerId.isAncillary() && annotationsChanged) ||
+                        (!providerId.isAncillary() && (propertiesChanged || annotationsChanged))) {
+                    boolean replaced = devices.replace(deviceId, existingDevice, newDevice);
+                    verify(replaced, "Replacing devices cache failed. PID:%s [expected:%s, found:%s, new=%s]",
+                            providerId, existingDevice, devices.get(deviceId), newDevice);
+                    eventType.set(DEVICE_UPDATED);
+                }
+            }
+            return newDevice;
+        });
+        if (eventType.get() != null && !providerId.isAncillary()) {
+            markOnline(deviceId);
+        }
+        return eventType.get() != null ? new DeviceEvent(eventType.get(), device) : null;
+    }
+
+    /**
+     * Returns the primary providerId for a device.
+     * @param deviceId device identifier
+     * @return primary providerId
+     */
+    private Set<ProviderId> getAllProviders(DeviceId deviceId) {
+        return deviceDescriptions.keySet()
+                                 .stream()
+                                 .filter(deviceKey -> deviceKey.deviceId().equals(deviceId))
+                                 .map(deviceKey -> deviceKey.providerId())
+                                 .collect(Collectors.toSet());
+    }
+
+    /**
+     * Returns the identifier for all providers for a device.
+     * @param deviceId device identifier
+     * @return set of provider identifiers
+     */
+    private ProviderId getPrimaryProviderId(DeviceId deviceId) {
+        Set<ProviderId> allProviderIds = getAllProviders(deviceId);
+        return allProviderIds.stream()
+                             .filter(p -> !p.isAncillary())
+                             .findFirst()
+                             .orElse(Iterables.getFirst(allProviderIds, null));
+    }
+
+    /**
+     * Returns a Device, merging descriptions from multiple Providers.
+     *
+     * @param deviceId      device identifier
+     * @return Device instance
+     */
+    private Device composeDevice(DeviceId deviceId) {
+
+        ProviderId primaryProviderId = getPrimaryProviderId(deviceId);
+        DeviceDescription primaryDeviceDescription =
+                deviceDescriptions.get(new DeviceKey(primaryProviderId, deviceId));
+
+        Type type = primaryDeviceDescription.type();
+        String manufacturer = primaryDeviceDescription.manufacturer();
+        String hwVersion = primaryDeviceDescription.hwVersion();
+        String swVersion = primaryDeviceDescription.swVersion();
+        String serialNumber = primaryDeviceDescription.serialNumber();
+        ChassisId chassisId = primaryDeviceDescription.chassisId();
+        DefaultAnnotations annotations = mergeAnnotations(deviceId);
+
+        return new DefaultDevice(primaryProviderId, deviceId, type, manufacturer,
+                                 hwVersion, swVersion, serialNumber,
+                                 chassisId, annotations);
+    }
+
+    private DeviceEvent purgeDeviceCache(DeviceId deviceId) {
+        Device removedDevice = devices.remove(deviceId);
+        if (removedDevice != null) {
+            getAllProviders(deviceId).forEach(p -> deviceDescriptions.remove(new DeviceKey(p, deviceId)));
+            return new DeviceEvent(DEVICE_REMOVED, removedDevice);
+        }
+        return null;
+    }
+
+    private boolean markOnline(DeviceId deviceId) {
+        return availableDevices.add(deviceId);
+    }
+
+    @Override
+    public DeviceEvent markOffline(DeviceId deviceId) {
+        availableDevices.remove(deviceId);
+        // set update listener will raise the event.
+        return null;
+    }
+
+    @Override
+    public List<DeviceEvent> updatePorts(ProviderId providerId,
+            DeviceId deviceId,
+            List<PortDescription> descriptions) {
+        NodeId master = mastershipService.getMasterFor(deviceId);
+        List<DeviceEvent> deviceEvents = null;
+        if (localNodeId.equals(master)) {
+            descriptions.forEach(description -> {
+                PortKey portKey = new PortKey(providerId, deviceId, description.portNumber());
+                portDescriptions.put(portKey, description);
+            });
+            deviceEvents = refreshDevicePortCache(providerId, deviceId, Optional.empty());
+        } else {
+            if (master == null) {
+                return Collections.emptyList();
+            }
+            PortInjectedEvent portInjectedEvent = new PortInjectedEvent(providerId, deviceId, descriptions);
+            deviceEvents = Futures.getUnchecked(
+                    clusterCommunicator.sendAndReceive(portInjectedEvent,
+                                    PORT_INJECTED,
+                                    SERIALIZER::encode,
+                                    SERIALIZER::decode,
+                                    master));
+        }
+        return deviceEvents == null ? Collections.emptyList() : deviceEvents;
+    }
+
+    private List<DeviceEvent> refreshDevicePortCache(ProviderId providerId,
+            DeviceId deviceId,
+            Optional<PortNumber> portNumber) {
+        Device device = devices.get(deviceId);
+        checkArgument(device != null, DEVICE_NOT_FOUND, deviceId);
+        List<DeviceEvent> events = Lists.newArrayList();
+
+        Map<PortNumber, Port> ports = devicePorts.computeIfAbsent(deviceId, key -> Maps.newConcurrentMap());
+        List<PortDescription> descriptions = Lists.newArrayList();
+        portDescriptions.entrySet().forEach(e -> {
+            PortKey key = e.getKey();
+            PortDescription value = e.getValue();
+            if (key.deviceId().equals(deviceId) && key.providerId().equals(providerId)) {
+                if (portNumber.isPresent()) {
+                    if (portNumber.get().equals(key.portNumber())) {
+                        descriptions.add(value);
+                    }
+                } else {
+                    descriptions.add(value);
+                }
+            }
+        });
+
+        for (PortDescription description : descriptions) {
+            final PortNumber number = description.portNumber();
+            ports.compute(number, (k, existingPort) -> {
+                Port newPort = composePort(device, number);
+                if (existingPort == null) {
+                    events.add(new DeviceEvent(PORT_ADDED, device, newPort));
+                } else {
+                    if (existingPort.isEnabled() != newPort.isEnabled() ||
+                            existingPort.type() != newPort.type() ||
+                            existingPort.portSpeed() != newPort.portSpeed() ||
+                            !AnnotationsUtil.isEqual(existingPort.annotations(), newPort.annotations())) {
+                        events.add(new DeviceEvent(PORT_UPDATED, device, newPort));
+                    }
+                }
+                return newPort;
+            });
+        }
+
+        return events;
+    }
+
+    /**
+     * Returns a Port, merging descriptions from multiple Providers.
+     *
+     * @param device   device the port is on
+     * @param number   port number
+     * @return Port instance
+     */
+    private Port composePort(Device device, PortNumber number) {
+
+        Map<ProviderId, PortDescription> descriptions = Maps.newHashMap();
+        portDescriptions.entrySet().forEach(entry -> {
+            PortKey portKey = entry.getKey();
+            if (portKey.deviceId().equals(device.id()) && portKey.portNumber().equals(number)) {
+                descriptions.put(portKey.providerId(), entry.getValue());
+            }
+        });
+        ProviderId primary = getPrimaryProviderId(device.id());
+        PortDescription primaryDescription = descriptions.get(primary);
+
+        // if no primary, assume not enabled
+        boolean isEnabled = false;
+        DefaultAnnotations annotations = DefaultAnnotations.builder().build();
+        if (primaryDescription != null) {
+            isEnabled = primaryDescription.isEnabled();
+            annotations = merge(annotations, primaryDescription.annotations());
+        }
+        Port updated = null;
+        for (Entry<ProviderId, PortDescription> e : descriptions.entrySet()) {
+            if (e.getKey().equals(primary)) {
+                continue;
+            }
+            annotations = merge(annotations, e.getValue().annotations());
+            updated = buildTypedPort(device, number, isEnabled, e.getValue(), annotations);
+        }
+        if (primaryDescription == null) {
+            return updated == null ? new DefaultPort(device, number, false, annotations) : updated;
+        }
+        return updated == null
+                ? buildTypedPort(device, number, isEnabled, primaryDescription, annotations)
+                : updated;
+    }
+
+    private Port buildTypedPort(Device device, PortNumber number, boolean isEnabled,
+            PortDescription description, Annotations annotations) {
+        switch (description.type()) {
+        case OMS:
+            OmsPortDescription omsDesc = (OmsPortDescription) description;
+            return new OmsPort(device, number, isEnabled, omsDesc.minFrequency(),
+                    omsDesc.maxFrequency(), omsDesc.grid(), annotations);
+        case OCH:
+            OchPortDescription ochDesc = (OchPortDescription) description;
+            return new OchPort(device, number, isEnabled, ochDesc.signalType(),
+                    ochDesc.isTunable(), ochDesc.lambda(), annotations);
+        case ODUCLT:
+            OduCltPortDescription oduDesc = (OduCltPortDescription) description;
+            return new OduCltPort(device, number, isEnabled, oduDesc.signalType(), annotations);
+        default:
+            return new DefaultPort(device, number, isEnabled, description.type(),
+                    description.portSpeed(), annotations);
+        }
+    }
+
+    @Override
+    public DeviceEvent updatePortStatus(ProviderId providerId,
+            DeviceId deviceId,
+            PortDescription portDescription) {
+        portDescriptions.put(new PortKey(providerId, deviceId, portDescription.portNumber()), portDescription);
+        List<DeviceEvent> events =
+                refreshDevicePortCache(providerId, deviceId, Optional.of(portDescription.portNumber()));
+        return Iterables.getFirst(events, null);
+    }
+
+    @Override
+    public List<Port> getPorts(DeviceId deviceId) {
+        return ImmutableList.copyOf(devicePorts.getOrDefault(deviceId, Maps.newHashMap()).values());
+    }
+
+    @Override
+    public Port getPort(DeviceId deviceId, PortNumber portNumber) {
+        return devicePorts.getOrDefault(deviceId, Maps.newHashMap()).get(portNumber);
+    }
+
+    @Override
+    public DeviceEvent updatePortStatistics(ProviderId providerId,
+            DeviceId deviceId,
+            Collection<PortStatistics> newStatsCollection) {
+
+        Map<PortNumber, PortStatistics> prvStatsMap = devicePortStats.get(deviceId);
+        Map<PortNumber, PortStatistics> newStatsMap = Maps.newHashMap();
+        Map<PortNumber, PortStatistics> deltaStatsMap = Maps.newHashMap();
+
+        if (prvStatsMap != null) {
+            for (PortStatistics newStats : newStatsCollection) {
+                PortNumber port = PortNumber.portNumber(newStats.port());
+                PortStatistics prvStats = prvStatsMap.get(port);
+                DefaultPortStatistics.Builder builder = DefaultPortStatistics.builder();
+                PortStatistics deltaStats = builder.build();
+                if (prvStats != null) {
+                    deltaStats = calcDeltaStats(deviceId, prvStats, newStats);
+                }
+                deltaStatsMap.put(port, deltaStats);
+                newStatsMap.put(port, newStats);
+            }
+        } else {
+            for (PortStatistics newStats : newStatsCollection) {
+                PortNumber port = PortNumber.portNumber(newStats.port());
+                newStatsMap.put(port, newStats);
+            }
+        }
+        devicePortDeltaStats.put(deviceId, deltaStatsMap);
+        devicePortStats.put(deviceId, newStatsMap);
+        // DeviceEvent returns null because of InternalPortStatsListener usage
+        return null;
+    }
+
+    /**
+     * Calculate delta statistics by subtracting previous from new statistics.
+     *
+     * @param deviceId
+     * @param prvStats
+     * @param newStats
+     * @return PortStatistics
+     */
+    public PortStatistics calcDeltaStats(DeviceId deviceId, PortStatistics prvStats, PortStatistics newStats) {
+        // calculate time difference
+        long deltaStatsSec, deltaStatsNano;
+        if (newStats.durationNano() < prvStats.durationNano()) {
+            deltaStatsNano = newStats.durationNano() - prvStats.durationNano() + TimeUnit.SECONDS.toNanos(1);
+            deltaStatsSec = newStats.durationSec() - prvStats.durationSec() - 1L;
+        } else {
+            deltaStatsNano = newStats.durationNano() - prvStats.durationNano();
+            deltaStatsSec = newStats.durationSec() - prvStats.durationSec();
+        }
+        DefaultPortStatistics.Builder builder = DefaultPortStatistics.builder();
+        DefaultPortStatistics deltaStats = builder.setDeviceId(deviceId)
+                .setPort(newStats.port())
+                .setPacketsReceived(newStats.packetsReceived() - prvStats.packetsReceived())
+                .setPacketsSent(newStats.packetsSent() - prvStats.packetsSent())
+                .setBytesReceived(newStats.bytesReceived() - prvStats.bytesReceived())
+                .setBytesSent(newStats.bytesSent() - prvStats.bytesSent())
+                .setPacketsRxDropped(newStats.packetsRxDropped() - prvStats.packetsRxDropped())
+                .setPacketsTxDropped(newStats.packetsTxDropped() - prvStats.packetsTxDropped())
+                .setPacketsRxErrors(newStats.packetsRxErrors() - prvStats.packetsRxErrors())
+                .setPacketsTxErrors(newStats.packetsTxErrors() - prvStats.packetsTxErrors())
+                .setDurationSec(deltaStatsSec)
+                .setDurationNano(deltaStatsNano)
+                .build();
+        return deltaStats;
+    }
+
+    @Override
+    public List<PortStatistics> getPortStatistics(DeviceId deviceId) {
+        Map<PortNumber, PortStatistics> portStats = devicePortStats.get(deviceId);
+        if (portStats == null) {
+            return Collections.emptyList();
+        }
+        return ImmutableList.copyOf(portStats.values());
+    }
+
+    @Override
+    public List<PortStatistics> getPortDeltaStatistics(DeviceId deviceId) {
+        Map<PortNumber, PortStatistics> portStats = devicePortDeltaStats.get(deviceId);
+        if (portStats == null) {
+            return Collections.emptyList();
+        }
+        return ImmutableList.copyOf(portStats.values());
+    }
+
+    @Override
+    public boolean isAvailable(DeviceId deviceId) {
+        return availableDevices.contains(deviceId);
+    }
+
+    @Override
+    public Iterable<Device> getAvailableDevices() {
+        return Iterables.filter(Iterables.transform(availableDevices, devices::get), d -> d != null);
+    }
+
+    @Override
+    public DeviceEvent removeDevice(DeviceId deviceId) {
+        NodeId master = mastershipService.getMasterFor(deviceId);
+        // if there exist a master, forward
+        // if there is no master, try to become one and process
+        boolean relinquishAtEnd = false;
+        if (master == null) {
+            final MastershipRole myRole = mastershipService.getLocalRole(deviceId);
+            if (myRole != MastershipRole.NONE) {
+                relinquishAtEnd = true;
+            }
+            log.debug("Temporarily requesting role for {} to remove", deviceId);
+            MastershipRole role = Futures.getUnchecked(mastershipService.requestRoleFor(deviceId));
+            if (role == MastershipRole.MASTER) {
+                master = localNodeId;
+            }
+        }
+
+        if (!localNodeId.equals(master)) {
+            log.debug("{} has control of {}, forwarding remove request",
+                      master, deviceId);
+
+            clusterCommunicator.unicast(deviceId, DEVICE_REMOVE_REQ, SERIALIZER::encode, master)
+                               .whenComplete((r, e) -> {
+                                   if (e != null) {
+                                       log.error("Failed to forward {} remove request to its master", deviceId, e);
+                                   }
+                               });
+            return null;
+        }
+
+        // I have control..
+        DeviceEvent event = null;
+        final DeviceKey deviceKey = new DeviceKey(getPrimaryProviderId(deviceId), deviceId);
+        DeviceDescription removedDeviceDescription =
+                deviceDescriptions.remove(deviceKey);
+        if (removedDeviceDescription != null) {
+            event = purgeDeviceCache(deviceId);
+        }
+
+        if (relinquishAtEnd) {
+            log.debug("Relinquishing temporary role acquired for {}", deviceId);
+            mastershipService.relinquishMastership(deviceId);
+        }
+        return event;
+    }
+
+    private DeviceEvent injectDevice(DeviceInjectedEvent event) {
+        return createOrUpdateDevice(event.providerId(), event.deviceId(), event.deviceDescription());
+    }
+
+    private List<DeviceEvent> injectPort(PortInjectedEvent event) {
+        return updatePorts(event.providerId(), event.deviceId(), event.portDescriptions());
+    }
+
+    private DefaultAnnotations mergeAnnotations(DeviceId deviceId) {
+        ProviderId primaryProviderId = getPrimaryProviderId(deviceId);
+        DeviceDescription primaryDeviceDescription =
+                deviceDescriptions.get(new DeviceKey(primaryProviderId, deviceId));
+        DefaultAnnotations annotations = DefaultAnnotations.builder().build();
+        annotations = merge(annotations, primaryDeviceDescription.annotations());
+        for (ProviderId providerId : getAllProviders(deviceId)) {
+            if (!providerId.equals(primaryProviderId)) {
+                annotations = merge(annotations,
+                                    deviceDescriptions.get(new DeviceKey(providerId, deviceId)).annotations());
+            }
+        }
+        return annotations;
+    }
+
+    private class InternalDeviceStatusTracker implements SetEventListener<DeviceId> {
+        @Override
+        public void event(SetEvent<DeviceId> event) {
+            final DeviceId deviceId = event.entry();
+            final Device device = devices.get(deviceId);
+            if (device != null) {
+                notifyDelegate(new DeviceEvent(DEVICE_AVAILABILITY_CHANGED, device));
+            } else {
+                pendingAvailableChangeUpdates.add(deviceId);
+            }
+        }
+    }
+
+    private class InternalDeviceChangeEventListener
+        implements EventuallyConsistentMapListener<DeviceKey, DeviceDescription> {
+        @Override
+        public void event(EventuallyConsistentMapEvent<DeviceKey, DeviceDescription> event) {
+            DeviceId deviceId = event.key().deviceId();
+            ProviderId providerId = event.key().providerId();
+            if (event.type() == PUT) {
+                notifyDelegate(refreshDeviceCache(providerId, deviceId));
+                if (pendingAvailableChangeUpdates.remove(deviceId)) {
+                    notifyDelegate(new DeviceEvent(DEVICE_AVAILABILITY_CHANGED, devices.get(deviceId)));
+                }
+            } else if (event.type() == REMOVE) {
+                notifyDelegate(purgeDeviceCache(deviceId));
+            }
+        }
+    }
+
+    private class InternalPortChangeEventListener
+        implements EventuallyConsistentMapListener<PortKey, PortDescription> {
+        @Override
+        public void event(EventuallyConsistentMapEvent<PortKey, PortDescription> event) {
+            DeviceId deviceId = event.key().deviceId();
+            ProviderId providerId = event.key().providerId();
+            PortNumber portNumber = event.key().portNumber();
+            if (event.type() == PUT) {
+                if (devices.containsKey(deviceId)) {
+                    List<DeviceEvent> events = refreshDevicePortCache(providerId, deviceId, Optional.of(portNumber));
+                    for (DeviceEvent deviceEvent : events) {
+                        notifyDelegate(deviceEvent);
+                    }
+                }
+            } else if (event.type() == REMOVE) {
+                log.warn("Unexpected port removed event");
+            }
+        }
+    }
+
+    private class InternalPortStatsListener
+        implements EventuallyConsistentMapListener<DeviceId, Map<PortNumber, PortStatistics>> {
+        @Override
+        public void event(EventuallyConsistentMapEvent<DeviceId, Map<PortNumber, PortStatistics>> event) {
+            if (event.type() == PUT) {
+                Device device = devices.get(event.key());
+                if (device != null) {
+                    delegate.notify(new DeviceEvent(PORT_STATS_UPDATED, device));
+                }
+            }
+        }
+    }
+}
\ No newline at end of file

+/*
+ * Copyright 2014-2015 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.device.impl;
+
+import java.util.Objects;
+
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.PortNumber;
+import org.onosproject.net.provider.ProviderId;
+
+import com.google.common.base.MoreObjects;
+
+/**
+ * Key for PortDescriptions in ECDeviceStore.
+ */
+public class PortKey {
+    private final ProviderId providerId;
+    private final DeviceId deviceId;
+    private final PortNumber portNumber;
+
+    public PortKey(ProviderId providerId, DeviceId deviceId, PortNumber portNumber) {
+        this.providerId = providerId;
+        this.deviceId = deviceId;
+        this.portNumber = portNumber;
+    }
+
+    public ProviderId providerId() {
+        return providerId;
+    }
+
+    public DeviceId deviceId() {
+        return deviceId;
+    }
+
+    public PortNumber portNumber() {
+        return portNumber;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(providerId, deviceId, portNumber);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (!(obj instanceof PortKey)) {
+            return false;
+        }
+        PortKey that = (PortKey) obj;
+        return Objects.equals(this.deviceId, that.deviceId) &&
+               Objects.equals(this.providerId, that.providerId) &&
+               Objects.equals(this.portNumber, that.portNumber);
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(getClass())
+                .add("providerId", providerId)
+                .add("deviceId", deviceId)
+                .add("portNumber", portNumber)
+                .toString();
+    }
+}