[ONOS-4260]Alarm and fault managment application refactoring according to ONOS architecture
Change-Id: I47e9db37eb5fc27ac19db2e4cb87774736b44685
diff --git a/apps/faultmanagement/fmmgr/src/main/java/org/onosproject/faultmanagement/impl/AlarmsManager.java b/apps/faultmanagement/fmmgr/src/main/java/org/onosproject/faultmanagement/impl/AlarmsManager.java
index 90ccaff..c7f724e 100644
--- a/apps/faultmanagement/fmmgr/src/main/java/org/onosproject/faultmanagement/impl/AlarmsManager.java
+++ b/apps/faultmanagement/fmmgr/src/main/java/org/onosproject/faultmanagement/impl/AlarmsManager.java
@@ -15,95 +15,79 @@
*/
package org.onosproject.faultmanagement.impl;
-import static com.google.common.base.Preconditions.checkNotNull;
-import java.util.Dictionary;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.stream.Collectors;
+import com.google.common.collect.ImmutableSet;
import org.apache.commons.collections.CollectionUtils;
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.Property;
-import static org.onlab.util.Tools.nullIsNotFound;
-
+import org.apache.felix.scr.annotations.Service;
+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.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;
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.slf4j.Logger;
-import static org.slf4j.LoggerFactory.getLogger;
-import org.apache.felix.scr.annotations.Service;
-import static org.onlab.util.Tools.get;
-import org.onosproject.core.ApplicationId;
-import org.onosproject.core.IdGenerator;
-import org.onosproject.core.CoreService;
-import org.apache.felix.scr.annotations.Reference;
-import org.apache.felix.scr.annotations.ReferenceCardinality;
-import org.onlab.util.ItemNotFoundException;
-import org.onosproject.incubator.net.faultmanagement.alarm.AlarmProvider;
-import org.onosproject.incubator.net.faultmanagement.alarm.DefaultAlarm;
-import org.onosproject.net.device.DeviceService;
+import org.onosproject.net.provider.AbstractProviderRegistry;
+import org.onosproject.net.provider.AbstractProviderService;
import org.osgi.service.component.ComponentContext;
-import static java.util.concurrent.TimeUnit.SECONDS;
+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 static org.onlab.util.Tools.groupedThreads;
-import org.onosproject.net.Device;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.onlab.util.Tools.nullIsNotFound;
+import static org.slf4j.LoggerFactory.getLogger;
/**
* Implementation of the Alarm service.
*/
@Component(immediate = true)
@Service
-public class AlarmsManager implements AlarmService {
-
- // For subscribing to device-related events
- @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
- protected CoreService coreService;
-
- @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
- protected DeviceService deviceService;
-
- @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
- protected AlarmProvider alarmProvider;
+public class AlarmsManager
+ extends AbstractProviderRegistry<AlarmProvider, AlarmProviderService>
+ implements AlarmService, AlarmProviderRegistry {
private final Logger log = getLogger(getClass());
- private ApplicationId appId;
- private IdGenerator idGenerator;
- private ScheduledExecutorService alarmPollExecutor;
-
- // dummy data
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<>();
+
+ private static final String NOT_SUPPORTED_YET = "Not supported yet.";
+
+ @Activate
+ public void activate(ComponentContext context) {
+ log.info("Started");
+ }
+
+ @Deactivate
+ public void deactivate(ComponentContext context) {
+ alarms.clear();
+ log.info("Stopped");
+ }
+
+ @Modified
+ public boolean modified(ComponentContext context) {
+ log.info("Modified");
+ return true;
+ }
+
private AlarmId generateAlarmId() {
return AlarmId.alarmId(alarmIdGenerator.incrementAndGet());
}
- private static final int DEFAULT_POLL_FREQUENCY_SECONDS = 120;
- @Property(name = "alarmPollFrequencySeconds", intValue = DEFAULT_POLL_FREQUENCY_SECONDS,
- label = "Frequency (in seconds) for polling alarm from devices")
- private int alarmPollFrequencySeconds = DEFAULT_POLL_FREQUENCY_SECONDS;
-
- // TODO implement purging of old alarms.
- private static final int DEFAULT_CLEAR_FREQUENCY_SECONDS = 500;
- @Property(name = "clearedAlarmPurgeSeconds", intValue = DEFAULT_CLEAR_FREQUENCY_SECONDS,
- label = "Frequency (in seconds) for deleting cleared alarms")
- private int clearedAlarmPurgeFrequencySeconds = DEFAULT_CLEAR_FREQUENCY_SECONDS;
-
- // TODO Later should must be persisted to disk or database
- private final Map<AlarmId, Alarm> alarms = new ConcurrentHashMap<>();
-
@Override
public Alarm updateBookkeepingFields(AlarmId id, boolean isAcknowledged, String assignedUser) {
@@ -120,7 +104,6 @@
}
public Alarm clear(AlarmId id) {
-
Alarm found = alarms.get(id);
if (found == null) {
log.warn("id {} cant be cleared as it is already gone.", id);
@@ -133,33 +116,25 @@
@Override
public Map<Alarm.SeverityLevel, Long> getAlarmCounts(DeviceId deviceId) {
-
return getAlarms(deviceId).stream().collect(
Collectors.groupingBy(Alarm::severity, Collectors.counting()));
-
}
@Override
public Map<Alarm.SeverityLevel, Long> getAlarmCounts() {
-
return getAlarms().stream().collect(
Collectors.groupingBy(Alarm::severity, Collectors.counting()));
}
-
- private static final String NOT_SUPPORTED_YET = "Not supported yet.";
-
@Override
public Alarm getAlarm(AlarmId alarmId) {
- return nullIsNotFound(
- alarms.get(
- checkNotNull(alarmId, "Alarm Id cannot be null")),
- "Alarm is not found");
+ return nullIsNotFound(alarms.get(checkNotNull(alarmId, "Alarm Id cannot be null")),
+ "Alarm is not found");
}
@Override
public Set<Alarm> getAlarms() {
- return new HashSet<>(alarms.values());
+ return ImmutableSet.copyOf(alarms.values());
}
@Override
@@ -198,122 +173,21 @@
@Override
public Set<Alarm> getAlarmsForLink(ConnectPoint src, ConnectPoint dst) {
- //TODO
throw new UnsupportedOperationException(NOT_SUPPORTED_YET);
}
@Override
public Set<Alarm> getAlarmsForFlow(DeviceId deviceId, long flowId) {
- //TODO
throw new UnsupportedOperationException(NOT_SUPPORTED_YET);
}
- private final AlarmListener alarmListener = new InternalAlarmListener();
-
- private class InternalAlarmListener implements AlarmListener {
-
- @Override
- public void event(AlarmEvent event) {
- log.debug("AlarmsManager. InternalAlarmListener received {}", event);
- try {
-
- switch (event.type()) {
- case DEVICE_DISCOVERY:
- DeviceId deviceId = checkNotNull(event.getDeviceRefreshed(), "Listener cannot be null");
- log.info("New alarm set for {} received!", deviceId);
- updateAlarms(event.subject(), deviceId);
- break;
-
- case NOTIFICATION:
- throw new IllegalArgumentException(
- "Alarm Notifications (Traps) not expected or implemented yet. Received =" + event);
- default:
- break;
- }
- } catch (Exception e) {
- log.warn("Failed to process {}", event, e);
- }
- }
- }
-
- @Activate
- public void activate(ComponentContext context) {
- appId = coreService.registerApplication("org.onosproject.faultmanagement.alarms");
- idGenerator = coreService.getIdGenerator("alarm-ids");
- log.info("Started with appId={}", appId);
-
- alarmProvider.addAlarmListener(alarmListener);
-
- probeActiveDevices();
-
- boolean result = modified(context);
- log.info("modified result = {}", result);
-
- alarmPollExecutor = newSingleThreadScheduledExecutor(groupedThreads("onos/fm", "alarms-poll-%d"));
- alarmPollExecutor.scheduleAtFixedRate(new PollAlarmsTask(),
- alarmPollFrequencySeconds, alarmPollFrequencySeconds, SECONDS);
-
- }
-
- /**
- * Auxiliary task to keep alarms up to date. IN future release alarm-notifications will be used as an optimization
- * so we dont have to wait until polling to detect changes. Furthermore with simple polling flapping alarms may be
- * missed.
- */
- private final class PollAlarmsTask implements Runnable {
-
- @Override
- public void run() {
- if (Thread.currentThread().isInterrupted()) {
- log.info("Interrupted, quitting");
- return;
- }
- try {
- probeActiveDevices();
- } catch (RuntimeException e) {
- log.error("Exception thrown during alarm synchronization process", e);
- }
- }
- }
-
- private void probeActiveDevices() {
- Iterable<Device> devices = deviceService.getAvailableDevices();
- log.info("Refresh alarms for all available devices={} ...", devices);
- for (Device d : devices) {
- log.info("Lets tell alarm provider to refresh alarms for {} ...", d.id());
- alarmProvider.triggerProbe(d.id());
- }
- }
-
- @Deactivate
- public void deactivate(ComponentContext context) {
- log.info("Deactivate ...");
- alarmProvider.removeAlarmListener(alarmListener);
-
- if (alarmPollExecutor != null) {
- alarmPollExecutor.shutdownNow();
- }
- alarms.clear();
- log.info("Stopped");
- }
-
- @Modified
- public boolean modified(ComponentContext context) {
- log.info("context={}", context);
- if (context == null) {
- log.info("No configuration file");
- return false;
- }
- Dictionary<?, ?> properties = context.getProperties();
- String clearedAlarmPurgeSeconds = get(properties, "clearedAlarmPurgeSeconds");
-
- log.info("Settings: clearedAlarmPurgeSeconds={}", clearedAlarmPurgeSeconds);
-
- return true;
+ @Override
+ protected AlarmProviderService createProviderService(AlarmProvider provider) {
+ return new InternalAlarmProviderService(provider);
}
// Synchronised to prevent duplicate NE alarms being raised
- synchronized void updateAlarms(Set<Alarm> discoveredSet, DeviceId deviceId) {
+ protected synchronized void updateAlarms(DeviceId deviceId, Set<Alarm> discoveredSet) {
Set<Alarm> storedSet = getActiveAlarms(deviceId);
log.trace("currentNeAlarms={}. discoveredAlarms={}", storedSet, discoveredSet);
@@ -324,16 +198,31 @@
storedSet.stream().filter(
(stored) -> (!discoveredSet.contains(stored))).forEach((stored) -> {
- log.info("Alarm will be cleared as it is not on the element. Cleared alarm: {}.", stored);
- clear(stored.id());
- });
+ log.debug("Alarm will be cleared as it is not on the element. Cleared alarm: {}.", stored);
+ clear(stored.id());
+ });
discoveredSet.stream().filter(
(discovered) -> (!storedSet.contains(discovered))).forEach((discovered) -> {
- log.info("Alarm will be raised as it is missing. New alarm: {}.", discovered);
+ log.info("New alarm raised as it is missing. New alarm: {}.", discovered);
AlarmId id = generateAlarmId();
alarms.put(id, new DefaultAlarm.Builder(discovered).withId(id).build());
});
}
+ private class InternalAlarmProviderService
+ extends AbstractProviderService<AlarmProvider>
+ implements AlarmProviderService {
+
+ InternalAlarmProviderService(AlarmProvider provider) {
+ super(provider);
+
+
+ }
+
+ @Override
+ public void updateAlarmList(DeviceId deviceId, Collection<Alarm> alarms) {
+ updateAlarms(deviceId, ImmutableSet.copyOf(alarms));
+ }
+ }
}