[ONOS-4287] Persistent and distributed alarm store

Change-Id: I2fb0f5d84e563a53f036be012a8190d7df5869dc

@@ -69,5 +69,12 @@
             <classifier>tests</classifier>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.onosproject</groupId>
+            <artifactId>onos-core-common</artifactId>
+            <version>${project.version}</version>
+            <classifier>tests</classifier>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>

+/*
+ * Copyright 2016-present 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.faultmanagement.api;
+
+import org.onosproject.incubator.net.faultmanagement.alarm.Alarm;
+import org.onosproject.incubator.net.faultmanagement.alarm.AlarmEvent;
+import org.onosproject.incubator.net.faultmanagement.alarm.AlarmId;
+import org.onosproject.net.DeviceId;
+import org.onosproject.store.Store;
+
+import java.util.Collection;
+
+/**
+ * Manages inventory of alarms; not intended for direct use.
+ */
+public interface AlarmStore extends Store<AlarmEvent, AlarmStoreDelegate> {
+
+    /**
+     * Retrieves and alarm based on it's id.
+     *
+     * @param alarmId alarm identifier
+     * @return alarm
+     */
+    Alarm getAlarm(AlarmId alarmId);
+
+    /**
+     * Retrieves all alarms present in the system.
+     *
+     * @return alarms
+     */
+    Collection<Alarm> getAlarms();
+
+    /**
+     * Retrieves alarms for a device.
+     *
+     * @param deviceId device identifier
+     * @return alarms
+     */
+    Collection<Alarm> getAlarms(DeviceId deviceId);
+
+    /**
+     * Stores an alarm.
+     *
+     * @param alarm alarm
+     */
+    void setAlarm(Alarm alarm);
+
+    /**
+     * Removes an alarm.
+     *
+     * @param alarmId alarm
+     */
+    void removeAlarm(AlarmId alarmId);
+}

+/*
+ * Copyright 2016-present 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.faultmanagement.api;
+
+import org.onosproject.incubator.net.faultmanagement.alarm.AlarmEvent;
+import org.onosproject.store.StoreDelegate;
+
+/**
+ * Infrastructure alarm store delegate abstraction.
+ */
+public interface AlarmStoreDelegate extends StoreDelegate<AlarmEvent> {
+}

+/*
+ * Copyright 2016-present 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.
+ */
+
+/**
+ * Infrastructure alarm store & related services API definitions.
+ */
+package org.onosproject.faultmanagement.api;

@@ -21,11 +21,17 @@ 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.Modified;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
 import org.apache.felix.scr.annotations.Service;
 import org.onlab.util.ItemNotFoundException;
+import org.onosproject.faultmanagement.api.AlarmStore;
+import org.onosproject.faultmanagement.api.AlarmStoreDelegate;
 import org.onosproject.incubator.net.faultmanagement.alarm.Alarm;
 import org.onosproject.incubator.net.faultmanagement.alarm.AlarmEntityId;
+import org.onosproject.incubator.net.faultmanagement.alarm.AlarmEvent;
 import org.onosproject.incubator.net.faultmanagement.alarm.AlarmId;
+import org.onosproject.incubator.net.faultmanagement.alarm.AlarmListener;
 import org.onosproject.incubator.net.faultmanagement.alarm.AlarmProvider;
 import org.onosproject.incubator.net.faultmanagement.alarm.AlarmProviderRegistry;
 import org.onosproject.incubator.net.faultmanagement.alarm.AlarmProviderService;
@@ -33,15 +39,13 @@ import org.onosproject.incubator.net.faultmanagement.alarm.AlarmService;
 import org.onosproject.incubator.net.faultmanagement.alarm.DefaultAlarm;
 import org.onosproject.net.ConnectPoint;
 import org.onosproject.net.DeviceId;
-import org.onosproject.net.provider.AbstractProviderRegistry;
+import org.onosproject.net.provider.AbstractListenerProviderRegistry;
 import org.onosproject.net.provider.AbstractProviderService;
-import org.osgi.service.component.ComponentContext;
 import org.slf4j.Logger;
 
 import java.util.Collection;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 
@@ -54,63 +58,67 @@ import static org.slf4j.LoggerFactory.getLogger;
  */
 @Component(immediate = true)
 @Service
-public class AlarmsManager
-        extends AbstractProviderRegistry<AlarmProvider, AlarmProviderService>
+public class AlarmManager
+        extends AbstractListenerProviderRegistry<AlarmEvent, AlarmListener, AlarmProvider, AlarmProviderService>
         implements AlarmService, AlarmProviderRegistry {
 
     private final Logger log = getLogger(getClass());
 
-    private final AtomicLong alarmIdGenerator = new AtomicLong(0);
 
-    // TODO Later should must be persisted to disk or database
-    protected final Map<AlarmId, Alarm> alarms = new ConcurrentHashMap<>();
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected AlarmStore store;
+
+    protected AlarmStoreDelegate delegate = this::post;
+
+    //TODO improve implementation of AlarmId
+    private final AtomicLong alarmIdGenerator = new AtomicLong(0);
 
     private static final String NOT_SUPPORTED_YET = "Not supported yet.";
 
     @Activate
-    public void activate(ComponentContext context) {
+    public void activate() {
+        store.setDelegate(delegate);
+        eventDispatcher.addSink(AlarmEvent.class, listenerRegistry);
         log.info("Started");
     }
 
     @Deactivate
-    public void deactivate(ComponentContext context) {
-        alarms.clear();
+    public void deactivate() {
+        store.unsetDelegate(delegate);
+        eventDispatcher.removeSink(AlarmEvent.class);
         log.info("Stopped");
     }
 
     @Modified
-    public boolean modified(ComponentContext context) {
+    public boolean modified() {
         log.info("Modified");
         return true;
     }
 
-    private AlarmId generateAlarmId() {
-        return AlarmId.alarmId(alarmIdGenerator.incrementAndGet());
-    }
-
     @Override
     public Alarm updateBookkeepingFields(AlarmId id, boolean isAcknowledged, String assignedUser) {
 
-        Alarm found = alarms.get(id);
+        Alarm found = store.getAlarm(id);
         if (found == null) {
             throw new ItemNotFoundException("Alarm with id " + id + " found");
         }
 
-        Alarm updated = new DefaultAlarm.Builder(found).
-                withAcknowledged(isAcknowledged).
-                withAssignedUser(assignedUser).build();
-        alarms.put(id, updated);
+        Alarm updated = new DefaultAlarm.Builder(found)
+                .withId(found.id())
+                .withAcknowledged(isAcknowledged)
+                .withAssignedUser(assignedUser).build();
+        store.setAlarm(updated);
         return updated;
     }
 
     public Alarm clear(AlarmId id) {
-        Alarm found = alarms.get(id);
+        Alarm found = store.getAlarm(id);
         if (found == null) {
-            log.warn("id {} cant be cleared as it is already gone.", id);
+            log.warn("Alarm {} is not present", id);
             return null;
         }
-        Alarm updated = new DefaultAlarm.Builder(found).clear().build();
-        alarms.put(id, updated);
+        Alarm updated = new DefaultAlarm.Builder(found).withId(id).clear().build();
+        store.setAlarm(updated);
         return updated;
     }
 
@@ -128,47 +136,44 @@ public class AlarmsManager
 
     @Override
     public Alarm getAlarm(AlarmId alarmId) {
-        return nullIsNotFound(alarms.get(checkNotNull(alarmId, "Alarm Id cannot be null")),
+        return nullIsNotFound(store.getAlarm(checkNotNull(alarmId, "Alarm Id cannot be null")),
                               "Alarm is not found");
     }
 
     @Override
     public Set<Alarm> getAlarms() {
-        return ImmutableSet.copyOf(alarms.values());
+        return ImmutableSet.copyOf(store.getAlarms());
     }
 
     @Override
     public Set<Alarm> getActiveAlarms() {
-        return alarms.values().stream().filter(
+        return ImmutableSet.copyOf(store.getAlarms().stream().filter(
                 a -> !a.severity().equals(Alarm.SeverityLevel.CLEARED)).
-                collect(Collectors.toSet());
+                collect(Collectors.toSet()));
     }
 
     @Override
     public Set<Alarm> getAlarms(Alarm.SeverityLevel severity) {
-        return alarms.values().stream().filter(
+        return ImmutableSet.copyOf(store.getAlarms().stream().filter(
                 a -> a.severity().equals(severity)).
-                collect(Collectors.toSet());
+                collect(Collectors.toSet()));
     }
 
     @Override
     public Set<Alarm> getAlarms(DeviceId deviceId) {
-        return alarms.values().stream().filter(
-                a -> deviceId.equals(a.deviceId())).
-                collect(Collectors.toSet());
+        return ImmutableSet.copyOf(store.getAlarms(deviceId));
     }
 
     private Set<Alarm> getActiveAlarms(DeviceId deviceId) {
-        return getActiveAlarms().stream().filter(
+        return ImmutableSet.copyOf(getActiveAlarms().stream().filter(
                 a -> deviceId.equals(a.deviceId())).
-                collect(Collectors.toSet());
+                collect(Collectors.toSet()));
     }
 
     @Override
     public Set<Alarm> getAlarms(DeviceId deviceId, AlarmEntityId source) {
-        return getAlarms(deviceId).stream().filter(
-                a -> source.equals(a.source())
-        ).collect(Collectors.toSet());
+        return ImmutableSet.copyOf(getAlarms(deviceId).stream().filter(
+                a -> source.equals(a.source())).collect(Collectors.toSet()));
     }
 
     @Override
@@ -189,35 +194,37 @@ public class AlarmsManager
     // Synchronised to prevent duplicate NE alarms being raised
     protected synchronized void updateAlarms(DeviceId deviceId, Set<Alarm> discoveredSet) {
         Set<Alarm> storedSet = getActiveAlarms(deviceId);
-        log.trace("currentNeAlarms={}. discoveredAlarms={}", storedSet, discoveredSet);
+        log.debug("CurrentNeAlarms={}. DiscoveredAlarms={}", storedSet, discoveredSet);
 
         if (CollectionUtils.isEqualCollection(storedSet, discoveredSet)) {
-            log.debug("Alarm lists are equivalent so no update for {}.", deviceId);
+            log.debug("No update for {}.", deviceId);
             return;
         }
-
+        //TODO implement distinction between UPDATED and CLEARED ALARMS
         storedSet.stream().filter(
                 (stored) -> (!discoveredSet.contains(stored))).forEach((stored) -> {
-            log.debug("Alarm will be cleared as it is not on the element. Cleared alarm: {}.", stored);
+            log.debug("Alarm will be Cleared as it is not on the device. Cleared alarm: {}.", stored);
             clear(stored.id());
         });
 
         discoveredSet.stream().filter(
                 (discovered) -> (!storedSet.contains(discovered))).forEach((discovered) -> {
-            log.info("New alarm raised as it is missing. New alarm: {}.", discovered);
+            log.info("New alarm raised {}", discovered);
             AlarmId id = generateAlarmId();
-            alarms.put(id, new DefaultAlarm.Builder(discovered).withId(id).build());
+            store.setAlarm(new DefaultAlarm.Builder(discovered).withId(id).build());
         });
     }
 
-    private class InternalAlarmProviderService
-            extends AbstractProviderService<AlarmProvider>
+    //TODO improve implementation of AlarmId
+    private AlarmId generateAlarmId() {
+        return AlarmId.alarmId(alarmIdGenerator.incrementAndGet());
+    }
+
+    private class InternalAlarmProviderService extends AbstractProviderService<AlarmProvider>
             implements AlarmProviderService {
 
         InternalAlarmProviderService(AlarmProvider provider) {
             super(provider);
-
-
         }
 
         @Override

+/*
+ * Copyright 2016-present 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.faultmanagement.impl;
+
+import com.google.common.collect.ImmutableSet;
+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.Modified;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.apache.felix.scr.annotations.Service;
+import org.onosproject.faultmanagement.api.AlarmStore;
+import org.onosproject.faultmanagement.api.AlarmStoreDelegate;
+import org.onosproject.incubator.net.faultmanagement.alarm.Alarm;
+import org.onosproject.incubator.net.faultmanagement.alarm.AlarmEntityId;
+import org.onosproject.incubator.net.faultmanagement.alarm.AlarmEvent;
+import org.onosproject.incubator.net.faultmanagement.alarm.AlarmId;
+import org.onosproject.incubator.net.faultmanagement.alarm.DefaultAlarm;
+import org.onosproject.net.DeviceId;
+import org.onosproject.store.AbstractStore;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.MapEvent;
+import org.onosproject.store.service.MapEventListener;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.StorageService;
+import org.slf4j.Logger;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Manages information of alarms using gossip protocol to distribute
+ * information.
+ */
+@Component(immediate = true)
+@Service
+public class DistributedAlarmStore
+        extends AbstractStore<AlarmEvent, AlarmStoreDelegate>
+        implements AlarmStore {
+
+    private final Logger log = getLogger(getClass());
+    private ConsistentMap<AlarmId, Alarm> alarms;
+    private Map<AlarmId, Alarm> alarmsMap;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected StorageService storageService;
+
+    private final MapEventListener<AlarmId, Alarm> listener = new InternalListener();
+
+    @Activate
+    public void activate() {
+        log.info("Started");
+        alarms = storageService.<AlarmId, Alarm>consistentMapBuilder()
+                .withName("onos-alarm-table")
+                .withSerializer(Serializer.using(KryoNamespaces.API,
+                                                 Alarm.class,
+                                                 DefaultAlarm.class,
+                                                 AlarmId.class,
+                                                 AlarmEvent.Type.class,
+                                                 Alarm.SeverityLevel.class,
+                                                 AlarmEntityId.class))
+                .build();
+        alarms.addListener(listener);
+        alarmsMap = alarms.asJavaMap();
+    }
+
+    @Deactivate
+    public void deactivate() {
+        alarms.removeListener(listener);
+        log.info("Stopped");
+    }
+
+    @Modified
+    public boolean modified() {
+        log.info("Modified");
+        return true;
+    }
+
+    @Override
+    public Alarm getAlarm(AlarmId alarmId) {
+        return alarmsMap.get(alarmId);
+    }
+
+    @Override
+    public Collection<Alarm> getAlarms() {
+        return ImmutableSet.copyOf(alarmsMap.values());
+    }
+
+    @Override
+    public Collection<Alarm> getAlarms(DeviceId deviceId) {
+        //FIXME: this is expensive, need refactoring when core maps provide different indexes.
+        return ImmutableSet.copyOf(alarmsMap.values().stream()
+                .filter(alarm -> alarm.deviceId().equals(deviceId))
+                .collect(Collectors.toSet()));
+    }
+
+    @Override
+    public void setAlarm(Alarm alarm) {
+        alarms.put(alarm.id(), alarm);
+    }
+
+    @Override
+    public void removeAlarm(AlarmId alarmId) {
+        alarms.remove(alarmId);
+    }
+
+    //Event listener to notify delegates about Map events.
+    private class InternalListener implements MapEventListener<AlarmId, Alarm> {
+
+        @Override
+        public void event(MapEvent<AlarmId, Alarm> mapEvent) {
+            final AlarmEvent.Type type;
+            final Alarm alarm;
+            switch (mapEvent.type()) {
+                case INSERT:
+                    type = AlarmEvent.Type.CREATED;
+                    alarm = mapEvent.newValue().value();
+                    break;
+                case UPDATE:
+                    type = AlarmEvent.Type.CREATED;
+                    alarm = mapEvent.newValue().value();
+                    break;
+                case REMOVE:
+                    type = AlarmEvent.Type.REMOVED;
+                    alarm = mapEvent.oldValue().value();
+                    break;
+                default:
+                    throw new IllegalArgumentException("Wrong event type " + mapEvent.type());
+            }
+            notifyDelegate(new AlarmEvent(type, alarm));
+        }
+    }
+}

@@ -15,6 +15,6 @@
  */
 
 /**
- * Fault Management application implementation.
+ * Infrastructure alarm model & related services implementation.
  */
 package org.onosproject.faultmanagement.impl;

+/*
+ * Copyright 2015-present 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.faultmanagement.impl;
+
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.onlab.junit.TestTools;
+import org.onlab.junit.TestUtils;
+import org.onlab.util.ItemNotFoundException;
+import org.onosproject.common.event.impl.TestEventDispatcher;
+import org.onosproject.event.Event;
+import org.onosproject.incubator.net.faultmanagement.alarm.Alarm;
+import org.onosproject.incubator.net.faultmanagement.alarm.AlarmEntityId;
+import org.onosproject.incubator.net.faultmanagement.alarm.AlarmEvent;
+import org.onosproject.incubator.net.faultmanagement.alarm.AlarmId;
+import org.onosproject.incubator.net.faultmanagement.alarm.AlarmListener;
+import org.onosproject.incubator.net.faultmanagement.alarm.DefaultAlarm;
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.NetTestTools;
+import org.onosproject.store.service.TestStorageService;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static junit.framework.TestCase.assertFalse;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.onosproject.incubator.net.faultmanagement.alarm.Alarm.SeverityLevel.CLEARED;
+import static org.onosproject.incubator.net.faultmanagement.alarm.Alarm.SeverityLevel.CRITICAL;
+
+/**
+ * Alarm manager test suite.
+ */
+public class AlarmManagerTest {
+
+    private static final DeviceId DEVICE_ID = DeviceId.deviceId("foo:bar");
+    private static final DefaultAlarm ALARM_A = new DefaultAlarm.Builder(
+            DEVICE_ID, "aaa", Alarm.SeverityLevel.CRITICAL, 0).build();
+
+    private static final DefaultAlarm ALARM_A_WITHSRC = new DefaultAlarm.Builder(
+            ALARM_A).forSource(AlarmEntityId.alarmEntityId("port:foo")).build();
+
+    private static final DefaultAlarm ALARM_B = new DefaultAlarm.Builder(
+            DEVICE_ID, "bbb", Alarm.SeverityLevel.CRITICAL, 0).build();
+
+    private AlarmManager manager;
+    private DistributedAlarmStore alarmStore;
+    protected TestListener listener = new TestListener();
+
+    @Rule
+    public final ExpectedException exception = ExpectedException.none();
+
+    @Before
+    public void setUp() throws Exception {
+        alarmStore = new DistributedAlarmStore();
+        TestUtils.setField(alarmStore, "storageService", new TestStorageService());
+        alarmStore.activate();
+        manager = new AlarmManager();
+        manager.addListener(listener);
+        NetTestTools.injectEventDispatcher(manager, new TestEventDispatcher());
+        manager.store = alarmStore;
+        manager.activate();
+    }
+
+    @Test
+    public void deactivate() throws Exception {
+        manager.updateAlarms(DEVICE_ID, ImmutableSet.of(ALARM_B, ALARM_A));
+        verifyGettingSetsOfAlarms(manager, 2, 2);
+        alarmStore.deactivate();
+        manager.removeListener(listener);
+        manager.deactivate();
+        NetTestTools.injectEventDispatcher(manager, null);
+        assertFalse("Store should not have delegate", alarmStore.hasDelegate());
+    }
+
+    @Test
+    public void testGettersWhenNoAlarms() {
+
+        assertTrue("No alarms should be present", manager.getAlarms().isEmpty());
+        assertTrue("No active alarms should be present", manager.getActiveAlarms().isEmpty());
+        assertTrue("The map should be empty per unknown device",
+                   manager.getAlarmCounts(DeviceId.NONE).keySet().isEmpty());
+        assertTrue("The counts should be empty", manager.getAlarmCounts().keySet().isEmpty());
+
+        assertEquals("Incorrect number of alarms for unknown device",
+                     0, manager.getAlarms(DeviceId.NONE).size());
+        assertEquals("Incorrect number of major alarms for unknown device",
+                     0, manager.getAlarms(Alarm.SeverityLevel.MAJOR).size());
+
+        exception.expect(NullPointerException.class);
+        manager.getAlarm(null);
+
+        exception.expect(ItemNotFoundException.class);
+        manager.getAlarm(AlarmId.alarmId(1));
+    }
+
+    @Test
+    public void testAlarmUpdates() throws InterruptedException {
+
+        assertTrue("No alarms should be present", manager.getAlarms().isEmpty());
+        manager.updateAlarms(DEVICE_ID, ImmutableSet.of());
+        assertTrue("No alarms should be present", manager.getAlarms().isEmpty());
+        Map<Alarm.SeverityLevel, Long> zeroAlarms = new CountsMapBuilder().create();
+        assertEquals("No alarms count should be present", zeroAlarms, manager.getAlarmCounts());
+        assertEquals("No alarms count should be present", zeroAlarms, manager.getAlarmCounts(DEVICE_ID));
+
+        manager.updateAlarms(DEVICE_ID, ImmutableSet.of(ALARM_B, ALARM_A));
+        verifyGettingSetsOfAlarms(manager, 2, 2);
+        validateEvents(AlarmEvent.Type.CREATED, AlarmEvent.Type.CREATED);
+        Map<Alarm.SeverityLevel, Long> critical2 = new CountsMapBuilder().with(CRITICAL, 2L).create();
+        assertEquals("A critical should be present", critical2, manager.getAlarmCounts());
+        assertEquals("A critical should be present", critical2, manager.getAlarmCounts(DEVICE_ID));
+
+        manager.updateAlarms(DEVICE_ID, ImmutableSet.of(ALARM_A));
+        verifyGettingSetsOfAlarms(manager, 2, 1);
+        validateEvents(AlarmEvent.Type.CREATED);
+        Map<Alarm.SeverityLevel, Long> critical1cleared1 =
+                new CountsMapBuilder().with(CRITICAL, 1L).with(CLEARED, 1L).create();
+        assertEquals("A critical should be present and cleared", critical1cleared1,
+                     manager.getAlarmCounts());
+        assertEquals("A critical should be present and cleared", critical1cleared1,
+                     manager.getAlarmCounts(DEVICE_ID));
+
+        // No change map when same alarms sent
+        manager.updateAlarms(DEVICE_ID, ImmutableSet.of(ALARM_A));
+        verifyGettingSetsOfAlarms(manager, 2, 1);
+        validateEvents();
+        assertEquals("Map should not be changed for same alarm", critical1cleared1,
+                     manager.getAlarmCounts());
+        assertEquals("Map should not be changed for same alarm", critical1cleared1,
+                     manager.getAlarmCounts(DEVICE_ID));
+
+        manager.updateAlarms(DEVICE_ID, ImmutableSet.of(ALARM_A, ALARM_A_WITHSRC));
+        verifyGettingSetsOfAlarms(manager, 3, 2);
+        validateEvents(AlarmEvent.Type.CREATED);
+        Map<Alarm.SeverityLevel, Long> critical2cleared1 =
+                new CountsMapBuilder().with(CRITICAL, 2L).with(CLEARED, 1L).create();
+        assertEquals("A critical should be present", critical2cleared1, manager.getAlarmCounts());
+        assertEquals("A critical should be present", critical2cleared1, manager.getAlarmCounts(DEVICE_ID));
+
+        manager.updateAlarms(DEVICE_ID, ImmutableSet.of());
+        verifyGettingSetsOfAlarms(manager, 3, 0);
+        validateEvents(AlarmEvent.Type.CREATED, AlarmEvent.Type.CREATED);
+        assertEquals(new CountsMapBuilder().with(CLEARED, 3L).create(), manager.getAlarmCounts(DEVICE_ID));
+
+        assertEquals("The counts should be empty for unknown devices", zeroAlarms,
+                     manager.getAlarmCounts(DeviceId.NONE));
+        assertEquals("The counts should be empty for unknown devices", zeroAlarms,
+                     manager.getAlarmCounts(DeviceId.deviceId("junk:junk")));
+
+    }
+
+    private void verifyGettingSetsOfAlarms(AlarmManager am, int expectedTotal, int expectedActive) {
+        assertEquals("Incorrect total alarms", expectedTotal, am.getAlarms().size());
+        assertEquals("Incorrect active alarms count", expectedActive, am.getActiveAlarms().size());
+    }
+
+    /**
+     * Method to validate that actual versus expected device key events were
+     * received correctly.
+     *
+     * @param types expected device key events.
+     */
+    private void validateEvents(Enum... types) {
+        TestTools.assertAfter(100, () -> {
+            int i = 0;
+            assertEquals("wrong events received", types.length, listener.events.size());
+            for (Event event : listener.events) {
+                assertEquals("incorrect event type", types[i], event.type());
+                i++;
+            }
+            listener.events.clear();
+        });
+    }
+
+    private static class CountsMapBuilder {
+
+        private final Map<Alarm.SeverityLevel, Long> map = new HashMap<>();
+
+        public CountsMapBuilder with(Alarm.SeverityLevel sev, Long count) {
+            map.put(sev, count);
+            return this;
+        }
+
+        public Map<Alarm.SeverityLevel, Long> create() {
+            return Collections.unmodifiableMap(map);
+        }
+    }
+
+
+    /**
+     * Test listener class to receive alarm events.
+     */
+    private static class TestListener implements AlarmListener {
+
+        protected List<AlarmEvent> events = Lists.newArrayList();
+
+        @Override
+        public void event(AlarmEvent event) {
+            events.add(event);
+        }
+
+    }
+}

-/*
- * Copyright 2015-present 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.faultmanagement.impl;
-
-import com.google.common.collect.ImmutableSet;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
-import org.onlab.util.ItemNotFoundException;
-import org.onosproject.incubator.net.faultmanagement.alarm.Alarm;
-import org.onosproject.incubator.net.faultmanagement.alarm.AlarmEntityId;
-import org.onosproject.incubator.net.faultmanagement.alarm.AlarmId;
-import org.onosproject.incubator.net.faultmanagement.alarm.DefaultAlarm;
-import org.onosproject.net.DeviceId;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-import static org.onosproject.incubator.net.faultmanagement.alarm.Alarm.SeverityLevel.CLEARED;
-import static org.onosproject.incubator.net.faultmanagement.alarm.Alarm.SeverityLevel.CRITICAL;
-
-/**
- * Alarm manager test suite.
- */
-public class AlarmsManagerTest {
-
-    private static final DeviceId DEVICE_ID = DeviceId.deviceId("foo:bar");
-    private static final DefaultAlarm ALARM_A = new DefaultAlarm.Builder(
-            DEVICE_ID, "aaa", Alarm.SeverityLevel.CRITICAL, 0).build();
-
-    private static final DefaultAlarm ALARM_A_WITHSRC = new DefaultAlarm.Builder(
-            ALARM_A).forSource(AlarmEntityId.alarmEntityId("port:foo")).build();
-
-    private static final DefaultAlarm ALARM_B = new DefaultAlarm.Builder(
-            DEVICE_ID, "bbb", Alarm.SeverityLevel.CRITICAL, 0).build();
-
-    private AlarmsManager am;
-
-    @Rule
-    public final ExpectedException exception = ExpectedException.none();
-
-    @Before
-    public void setUp() throws Exception {
-        am = new AlarmsManager();
-    }
-
-    @Test
-    public void deactivate() throws Exception {
-        am.updateAlarms(DEVICE_ID, ImmutableSet.of(ALARM_B, ALARM_A));
-        verifyGettingSetsOfAlarms(am, 2, 2);
-        am.deactivate(null);
-        assertEquals("Alarms should be purged", 0, am.alarms.size());
-    }
-
-    @Test
-    public void testGettersWhenNoAlarms() {
-
-        assertTrue("No alarms should be present", am.getAlarms().isEmpty());
-        assertTrue("No active alarms should be present", am.getActiveAlarms().isEmpty());
-        assertTrue("The map should be empty per unknown device",
-                   am.getAlarmCounts(DeviceId.NONE).keySet().isEmpty());
-        assertTrue("The counts should be empty", am.getAlarmCounts().keySet().isEmpty());
-
-        assertEquals("Incorrect number of alarms for unknown device",
-                     0, am.getAlarms(DeviceId.NONE).size());
-        assertEquals("Incorrect number of major alarms for unknown device",
-                     0, am.getAlarms(Alarm.SeverityLevel.MAJOR).size());
-
-        exception.expect(NullPointerException.class);
-        am.getAlarm(null);
-
-        exception.expect(ItemNotFoundException.class);
-        am.getAlarm(AlarmId.alarmId(1));
-    }
-
-    @Test
-    public void testAlarmUpdates() {
-
-        assertTrue("No alarms should be present", am.getAlarms().isEmpty());
-        am.updateAlarms(DEVICE_ID, ImmutableSet.of());
-        assertTrue("No alarms should be present", am.getAlarms().isEmpty());
-        Map<Alarm.SeverityLevel, Long> zeroAlarms = new CountsMapBuilder().create();
-        assertEquals("No alarms count should be present", zeroAlarms, am.getAlarmCounts());
-        assertEquals("No alarms count should be present", zeroAlarms, am.getAlarmCounts(DEVICE_ID));
-
-        am.updateAlarms(DEVICE_ID, ImmutableSet.of(ALARM_B, ALARM_A));
-        verifyGettingSetsOfAlarms(am, 2, 2);
-        Map<Alarm.SeverityLevel, Long> critical2 = new CountsMapBuilder().with(CRITICAL, 2L).create();
-        assertEquals("A critical should be present", critical2, am.getAlarmCounts());
-        assertEquals("A critical should be present", critical2, am.getAlarmCounts(DEVICE_ID));
-
-        am.updateAlarms(DEVICE_ID, ImmutableSet.of(ALARM_A));
-        verifyGettingSetsOfAlarms(am, 2, 1);
-        Map<Alarm.SeverityLevel, Long> critical1cleared1 =
-                new CountsMapBuilder().with(CRITICAL, 1L).with(CLEARED, 1L).create();
-        assertEquals("A critical should be present and cleared", critical1cleared1,
-                     am.getAlarmCounts());
-        assertEquals("A critical should be present and cleared", critical1cleared1,
-                     am.getAlarmCounts(DEVICE_ID));
-
-        // No change map when same alarms sent
-        am.updateAlarms(DEVICE_ID, ImmutableSet.of(ALARM_A));
-        verifyGettingSetsOfAlarms(am, 2, 1);
-        assertEquals("Map should not be changed for same alarm", critical1cleared1,
-                     am.getAlarmCounts());
-        assertEquals("Map should not be changed for same alarm", critical1cleared1,
-                     am.getAlarmCounts(DEVICE_ID));
-
-        am.updateAlarms(DEVICE_ID, ImmutableSet.of(ALARM_A, ALARM_A_WITHSRC));
-        verifyGettingSetsOfAlarms(am, 3, 2);
-        Map<Alarm.SeverityLevel, Long> critical2cleared1 =
-                new CountsMapBuilder().with(CRITICAL, 2L).with(CLEARED, 1L).create();
-        assertEquals("A critical should be present", critical2cleared1, am.getAlarmCounts());
-        assertEquals("A critical should be present", critical2cleared1, am.getAlarmCounts(DEVICE_ID));
-
-        am.updateAlarms(DEVICE_ID, ImmutableSet.of());
-        verifyGettingSetsOfAlarms(am, 3, 0);
-        assertEquals(new CountsMapBuilder().with(CLEARED, 3L).create(), am.getAlarmCounts(DEVICE_ID));
-
-        assertEquals("The counts should be empty for unknown devices", zeroAlarms,
-                     am.getAlarmCounts(DeviceId.NONE));
-        assertEquals("The counts should be empty for unknown devices", zeroAlarms,
-                     am.getAlarmCounts(DeviceId.deviceId("junk:junk")));
-
-    }
-
-    private void verifyGettingSetsOfAlarms(AlarmsManager am, int expectedTotal, int expectedActive) {
-        assertEquals("Incorrect total alarms", expectedTotal, am.getAlarms().size());
-        assertEquals("Incorrect active alarms count", expectedActive, am.getActiveAlarms().size());
-    }
-
-    private static class CountsMapBuilder {
-
-        private final Map<Alarm.SeverityLevel, Long> map = new HashMap<>();
-
-        public CountsMapBuilder with(Alarm.SeverityLevel sev, Long count) {
-            map.put(sev, count);
-            return this;
-        }
-
-        public Map<Alarm.SeverityLevel, Long> create() {
-            return Collections.unmodifiableMap(map);
-        }
-    }
-
-}

+/*
+ * Copyright 2016-present 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.faultmanagement.impl;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.onosproject.incubator.net.faultmanagement.alarm.Alarm;
+import org.onosproject.incubator.net.faultmanagement.alarm.DefaultAlarm;
+import org.onosproject.net.DeviceId;
+import org.onosproject.store.service.TestStorageService;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Distributed Alarm store test suite.
+ */
+public class DistributedAlarmStoreTest {
+    private DistributedAlarmStore alarmStore;
+    private static final DeviceId DEVICE_ID = DeviceId.deviceId("foo:bar");
+    private static final DefaultAlarm ALARM_A = new DefaultAlarm.Builder(
+            DEVICE_ID, "aaa", Alarm.SeverityLevel.CRITICAL, 0).build();
+
+    /**
+     * Sets up the device key store and the storage service test harness.
+     */
+    @Before
+    public void setUp() {
+        alarmStore = new DistributedAlarmStore();
+        alarmStore.storageService = new TestStorageService();
+        alarmStore.setDelegate(event -> {
+        });
+        alarmStore.activate();
+    }
+
+    /**
+     * Tears down the device key store.
+     */
+    @After
+    public void tearDown() {
+        alarmStore.deactivate();
+    }
+
+    /**
+     * Tests adding, removing and getting.
+     */
+    @Test
+    public void basics() {
+        alarmStore.setAlarm(ALARM_A);
+        assertTrue("There should be one alarm in the set.",
+                   alarmStore.getAlarms().contains(ALARM_A));
+        assertTrue("The same alarm should be returned.",
+                   alarmStore.getAlarms(DEVICE_ID).contains(ALARM_A));
+        assertTrue("The alarm should be the same",
+                   alarmStore.getAlarm(ALARM_A.id()).equals(ALARM_A));
+        alarmStore.removeAlarm(ALARM_A.id());
+        assertFalse("There should be no alarm in the set.",
+                    alarmStore.getAlarms().contains(ALARM_A));
+    }
+
+}

@@ -15,64 +15,36 @@
  */
 package org.onosproject.incubator.net.faultmanagement.alarm;
 
-import java.util.Set;
 import org.onosproject.event.AbstractEvent;
-import org.onosproject.net.DeviceId;
 
 /**
- * Entity that represents Alarm events. Note: although the event will itself have a time, consumers may be more
- * interested in the times embedded in the alarms themselves.
- *
+ * Entity that represents Alarm events. Note: although the event will itself have a time,
+ * consumers may be more interested in the times embedded in the alarms themselves.
  */
-public class AlarmEvent extends AbstractEvent<AlarmEvent.Type, Set<Alarm>> {
-
-    private final DeviceId deviceRefreshed;
+public class AlarmEvent extends AbstractEvent<AlarmEvent.Type, Alarm> {
 
     /**
-     * Creates an event due to one or more notification.
-     *
-     * @param alarms the set one or more of alarms.
+     * Type of alarm event.
      */
-    public AlarmEvent(Set<Alarm> alarms) {
-        super(Type.NOTIFICATION, alarms);
-        deviceRefreshed = null;
-    }
+    public enum Type {
 
         /**
-     * Creates an event due to alarm discovery for a device.
-     *
-     * @param alarms the set of alarms.
-     * @param deviceRefreshed if of refreshed device, populated after a de-discovery
+         * Individual alarm updated.
          */
-    public AlarmEvent(Set<Alarm> alarms,
-            DeviceId deviceRefreshed) {
-        super(Type.DEVICE_DISCOVERY, alarms);
-        this.deviceRefreshed = deviceRefreshed;
-
-    }
-
+        CREATED,
         /**
-     * Gets which device was refreshed.
-     *
-     * @return the refreshed device, or null if event related to a asynchronous notification(s)
+         * Alarm set updated for a given device.
          */
-    public DeviceId getDeviceRefreshed() {
-        return deviceRefreshed;
+        REMOVED,
     }
 
     /**
-     * Type of alarm event.
-     */
-    public enum Type {
-
-        /**
-         * Individual alarm(s) updated.
-         */
-        NOTIFICATION,
-        /**
-         * Alarm set updated for a given device.
+     * Creates an event due to one alarm.
+     *
+     * @param alarm the alarm related to the event.
      */
-        DEVICE_DISCOVERY,
+    public AlarmEvent(AlarmEvent.Type type, Alarm alarm) {
+        super(type, alarm);
     }
 
 }

+/*
+ * Copyright 2016-present 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.incubator.net.faultmanagement.alarm;
+
+import org.onosproject.event.EventListener;
+
+/**
+ * Entity capable of receiving alarm related events.
+ */
+public interface AlarmListener extends EventListener<AlarmEvent> {
+}
\ No newline at end of file

@@ -16,7 +16,6 @@
 package org.onosproject.incubator.net.faultmanagement.alarm;
 
 
-import com.google.common.annotations.Beta;
 import org.onosproject.net.DeviceId;
 import org.onosproject.net.provider.ProviderService;
 
@@ -25,7 +24,7 @@ import java.util.Collection;
 /**
  * The interface Alarm provider service.
  */
-@Beta
+
 public interface AlarmProviderService extends ProviderService<AlarmProvider> {
 
     /**

@@ -15,19 +15,18 @@
  */
 package org.onosproject.incubator.net.faultmanagement.alarm;
 
-import com.google.common.annotations.Beta;
-import java.util.Map;
-
-import java.util.Set;
+import org.onosproject.event.ListenerService;
 import org.onosproject.net.ConnectPoint;
 import org.onosproject.net.DeviceId;
 
+import java.util.Map;
+import java.util.Set;
+
 /**
  * Service for interacting with the alarm handling of devices. Unless stated otherwise, getter methods
  * return active AND recently-cleared alarms.
  */
-@Beta
-public interface AlarmService {
+public interface AlarmService extends ListenerService<AlarmEvent, AlarmListener> {
 
     /**
      * Update book-keeping (ie administrative) fields for the alarm matching the specified identifier.
@@ -36,7 +35,6 @@ public interface AlarmService {
      * @param isAcknowledged new acknowledged state
      * @param assignedUser   new assigned user, null clear
      * @return updated alarm (including any recent device-derived changes)
-     *
      */
     Alarm updateBookkeepingFields(AlarmId id, boolean isAcknowledged, String assignedUser);
 

@@ -25,6 +25,7 @@ import static com.google.common.base.Preconditions.checkNotNull;
 /**
  * Default implementation of an alarm.
  */
+//TODO simpler creation and updating.
 public final class DefaultAlarm implements Alarm {
 
     private final AlarmId id;
@@ -41,6 +42,22 @@ public final class DefaultAlarm implements Alarm {
     private final boolean isManuallyClearable;
     private final String assignedUser;
 
+    //Only for Kryo
+    DefaultAlarm() {
+        id = null;
+        deviceId = null;
+        description = null;
+        source = null;
+        timeRaised = -1;
+        timeUpdated = -1;
+        timeCleared = null;
+        severity = null;
+        isServiceAffecting = false;
+        isAcknowledged = false;
+        isManuallyClearable = false;
+        assignedUser = null;
+    }
+
     /**
      * Instantiates a new Default alarm.
      *