Brian O'Connor

Adding configurable variables to DistributedFlowRuleStore

Change-Id: I989de6b4ca4caf8ac1e648fbae766d0f88414419
...@@ -74,11 +74,6 @@ ...@@ -74,11 +74,6 @@
74 </dependency> 74 </dependency>
75 75
76 <dependency> 76 <dependency>
77 - <groupId>org.osgi</groupId>
78 - <artifactId>org.osgi.compendium</artifactId>
79 - </dependency>
80 -
81 - <dependency>
82 <groupId>org.apache.felix</groupId> 77 <groupId>org.apache.felix</groupId>
83 <artifactId>org.apache.felix.scr.annotations</artifactId> 78 <artifactId>org.apache.felix.scr.annotations</artifactId>
84 </dependency> 79 </dependency>
......
...@@ -52,6 +52,10 @@ ...@@ -52,6 +52,10 @@
52 <artifactId>onlab-junit</artifactId> 52 <artifactId>onlab-junit</artifactId>
53 <scope>test</scope> 53 <scope>test</scope>
54 </dependency> 54 </dependency>
55 + <dependency>
56 + <groupId>org.osgi</groupId>
57 + <artifactId>org.osgi.compendium</artifactId>
58 + </dependency>
55 </dependencies> 59 </dependencies>
56 60
57 <build> 61 <build>
......
...@@ -23,10 +23,13 @@ import org.apache.commons.lang.math.RandomUtils; ...@@ -23,10 +23,13 @@ import org.apache.commons.lang.math.RandomUtils;
23 import org.apache.felix.scr.annotations.Activate; 23 import org.apache.felix.scr.annotations.Activate;
24 import org.apache.felix.scr.annotations.Component; 24 import org.apache.felix.scr.annotations.Component;
25 import org.apache.felix.scr.annotations.Deactivate; 25 import org.apache.felix.scr.annotations.Deactivate;
26 +import org.apache.felix.scr.annotations.Modified;
27 +import org.apache.felix.scr.annotations.Property;
26 import org.apache.felix.scr.annotations.Reference; 28 import org.apache.felix.scr.annotations.Reference;
27 import org.apache.felix.scr.annotations.ReferenceCardinality; 29 import org.apache.felix.scr.annotations.ReferenceCardinality;
28 import org.apache.felix.scr.annotations.Service; 30 import org.apache.felix.scr.annotations.Service;
29 import org.onlab.util.KryoNamespace; 31 import org.onlab.util.KryoNamespace;
32 +import org.onosproject.cfg.ComponentConfigService;
30 import org.onosproject.cluster.ClusterService; 33 import org.onosproject.cluster.ClusterService;
31 import org.onosproject.cluster.NodeId; 34 import org.onosproject.cluster.NodeId;
32 import org.onosproject.core.CoreService; 35 import org.onosproject.core.CoreService;
...@@ -68,12 +71,14 @@ import org.onosproject.store.serializers.KryoNamespaces; ...@@ -68,12 +71,14 @@ import org.onosproject.store.serializers.KryoNamespaces;
68 import org.onosproject.store.serializers.KryoSerializer; 71 import org.onosproject.store.serializers.KryoSerializer;
69 import org.onosproject.store.serializers.StoreSerializer; 72 import org.onosproject.store.serializers.StoreSerializer;
70 import org.onosproject.store.serializers.impl.DistributedStoreSerializers; 73 import org.onosproject.store.serializers.impl.DistributedStoreSerializers;
74 +import org.osgi.service.component.ComponentContext;
71 import org.slf4j.Logger; 75 import org.slf4j.Logger;
72 76
73 import java.io.IOException; 77 import java.io.IOException;
74 import java.util.Arrays; 78 import java.util.Arrays;
75 import java.util.Collection; 79 import java.util.Collection;
76 import java.util.Collections; 80 import java.util.Collections;
81 +import java.util.Dictionary;
77 import java.util.List; 82 import java.util.List;
78 import java.util.Map; 83 import java.util.Map;
79 import java.util.Set; 84 import java.util.Set;
...@@ -88,6 +93,8 @@ import java.util.concurrent.TimeUnit; ...@@ -88,6 +93,8 @@ import java.util.concurrent.TimeUnit;
88 import java.util.concurrent.TimeoutException; 93 import java.util.concurrent.TimeoutException;
89 import java.util.stream.Collectors; 94 import java.util.stream.Collectors;
90 95
96 +import static com.google.common.base.Strings.isNullOrEmpty;
97 +import static org.apache.felix.scr.annotations.ReferenceCardinality.MANDATORY_UNARY;
91 import static org.onlab.util.Tools.groupedThreads; 98 import static org.onlab.util.Tools.groupedThreads;
92 import static org.onosproject.net.flow.FlowRuleEvent.Type.RULE_REMOVED; 99 import static org.onosproject.net.flow.FlowRuleEvent.Type.RULE_REMOVED;
93 import static org.onosproject.store.flow.impl.FlowStoreMessageSubjects.*; 100 import static org.onosproject.store.flow.impl.FlowStoreMessageSubjects.*;
...@@ -104,8 +111,17 @@ public class DistributedFlowRuleStore ...@@ -104,8 +111,17 @@ public class DistributedFlowRuleStore
104 111
105 private final Logger log = getLogger(getClass()); 112 private final Logger log = getLogger(getClass());
106 113
107 - // TODO: Make configurable.
108 private static final int MESSAGE_HANDLER_THREAD_POOL_SIZE = 8; 114 private static final int MESSAGE_HANDLER_THREAD_POOL_SIZE = 8;
115 + private static final boolean DEFAULT_BACKUP_ENABLED = false;
116 + private static final long FLOW_RULE_STORE_TIMEOUT_MILLIS = 5000;
117 +
118 + @Property(name = "msgHandlerPoolSize", intValue = MESSAGE_HANDLER_THREAD_POOL_SIZE,
119 + label = "Number of threads in the message handler pool")
120 + private int msgHandlerPoolSize = MESSAGE_HANDLER_THREAD_POOL_SIZE;
121 +
122 + @Property(name = "backupEnabled", boolValue = DEFAULT_BACKUP_ENABLED,
123 + label = "Indicates whether backups are enabled or not")
124 + private boolean backupEnabled = DEFAULT_BACKUP_ENABLED;
109 125
110 private InternalFlowTable flowTable; 126 private InternalFlowTable flowTable;
111 127
...@@ -127,6 +143,9 @@ public class DistributedFlowRuleStore ...@@ -127,6 +143,9 @@ public class DistributedFlowRuleStore
127 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY) 143 @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
128 protected CoreService coreService; 144 protected CoreService coreService;
129 145
146 + @Reference(cardinality = MANDATORY_UNARY)
147 + protected ComponentConfigService configService;
148 +
130 private Map<Long, NodeId> pendingResponses = Maps.newConcurrentMap(); 149 private Map<Long, NodeId> pendingResponses = Maps.newConcurrentMap();
131 150
132 private ExecutorService messageHandlingExecutor; 151 private ExecutorService messageHandlingExecutor;
...@@ -143,24 +162,22 @@ public class DistributedFlowRuleStore ...@@ -143,24 +162,22 @@ public class DistributedFlowRuleStore
143 } 162 }
144 }; 163 };
145 164
146 - private static final long FLOW_RULE_STORE_TIMEOUT_MILLIS = 5000;
147 -
148 private ReplicaInfoEventListener replicaInfoEventListener; 165 private ReplicaInfoEventListener replicaInfoEventListener;
149 166
150 private IdGenerator idGenerator; 167 private IdGenerator idGenerator;
151 168
152 @Activate 169 @Activate
153 - public void activate() { 170 + public void activate(ComponentContext context) {
171 + configService.registerProperties(getClass());
154 172
155 - flowTable = new InternalFlowTable(); // .withBackupsEnabled(false); 173 + flowTable = new InternalFlowTable().withBackupsEnabled(backupEnabled);
156 174
157 idGenerator = coreService.getIdGenerator(FlowRuleService.FLOW_OP_TOPIC); 175 idGenerator = coreService.getIdGenerator(FlowRuleService.FLOW_OP_TOPIC);
158 176
159 final NodeId local = clusterService.getLocalNode().id(); 177 final NodeId local = clusterService.getLocalNode().id();
160 178
161 messageHandlingExecutor = Executors.newFixedThreadPool( 179 messageHandlingExecutor = Executors.newFixedThreadPool(
162 - MESSAGE_HANDLER_THREAD_POOL_SIZE, 180 + msgHandlerPoolSize, groupedThreads("onos/store/flow", "message-handlers"));
163 - groupedThreads("onos/store/flow", "message-handlers"));
164 181
165 clusterCommunicator.addSubscriber(APPLY_BATCH_FLOWS, new OnStoreBatch(local), messageHandlingExecutor); 182 clusterCommunicator.addSubscriber(APPLY_BATCH_FLOWS, new OnStoreBatch(local), messageHandlingExecutor);
166 183
...@@ -222,11 +239,12 @@ public class DistributedFlowRuleStore ...@@ -222,11 +239,12 @@ public class DistributedFlowRuleStore
222 239
223 replicaInfoManager.addListener(replicaInfoEventListener); 240 replicaInfoManager.addListener(replicaInfoEventListener);
224 241
225 - log.info("Started"); 242 + logConfig("Started");
226 } 243 }
227 244
228 @Deactivate 245 @Deactivate
229 - public void deactivate() { 246 + public void deactivate(ComponentContext context) {
247 + configService.unregisterProperties(getClass(), false);
230 clusterCommunicator.removeSubscriber(REMOVE_FLOW_ENTRY); 248 clusterCommunicator.removeSubscriber(REMOVE_FLOW_ENTRY);
231 clusterCommunicator.removeSubscriber(GET_DEVICE_FLOW_ENTRIES); 249 clusterCommunicator.removeSubscriber(GET_DEVICE_FLOW_ENTRIES);
232 clusterCommunicator.removeSubscriber(GET_FLOW_ENTRY); 250 clusterCommunicator.removeSubscriber(GET_FLOW_ENTRY);
...@@ -237,6 +255,46 @@ public class DistributedFlowRuleStore ...@@ -237,6 +255,46 @@ public class DistributedFlowRuleStore
237 log.info("Stopped"); 255 log.info("Stopped");
238 } 256 }
239 257
258 + @Modified
259 + public void modified(ComponentContext context) {
260 + if (context == null) {
261 + backupEnabled = DEFAULT_BACKUP_ENABLED;
262 + logConfig("Default config");
263 + return;
264 + }
265 +
266 + Dictionary properties = context.getProperties();
267 + int newPoolSize;
268 + boolean newBackupEnabled;
269 + try {
270 + String s = (String) properties.get("msgHandlerPoolSize");
271 + newPoolSize = isNullOrEmpty(s) ? msgHandlerPoolSize : Integer.parseInt(s.trim());
272 +
273 + s = (String) properties.get("backupEnabled");
274 + newBackupEnabled = isNullOrEmpty(s) ? backupEnabled : Boolean.parseBoolean(s.trim());
275 +
276 + } catch (NumberFormatException | ClassCastException e) {
277 + newPoolSize = MESSAGE_HANDLER_THREAD_POOL_SIZE;
278 + newBackupEnabled = DEFAULT_BACKUP_ENABLED;
279 + }
280 +
281 + if (newPoolSize != msgHandlerPoolSize || newBackupEnabled != backupEnabled) {
282 + msgHandlerPoolSize = newPoolSize;
283 + backupEnabled = newBackupEnabled;
284 + // reconfigure the store
285 + flowTable.withBackupsEnabled(backupEnabled);
286 + ExecutorService oldMsgHandler = messageHandlingExecutor;
287 + messageHandlingExecutor = Executors.newFixedThreadPool(
288 + msgHandlerPoolSize, groupedThreads("onos/store/flow", "message-handlers"));
289 + oldMsgHandler.shutdown();
290 + logConfig("Reconfigured");
291 + }
292 + }
293 +
294 + private void logConfig(String prefix) {
295 + log.info("{} with msgHandlerPoolSize = {}; backupEnabled = {}",
296 + prefix, msgHandlerPoolSize, backupEnabled);
297 + }
240 298
241 // This is not a efficient operation on a distributed sharded 299 // This is not a efficient operation on a distributed sharded
242 // flow store. We need to revisit the need for this operation or at least 300 // flow store. We need to revisit the need for this operation or at least
...@@ -639,14 +697,13 @@ public class DistributedFlowRuleStore ...@@ -639,14 +697,13 @@ public class DistributedFlowRuleStore
639 (flowId, flowEntry) -> 697 (flowId, flowEntry) ->
640 (flowEntry == null) ? null : deviceClockService.getTimestamp(flowEntry.deviceId()); 698 (flowEntry == null) ? null : deviceClockService.getTimestamp(flowEntry.deviceId());
641 699
642 - private final EventuallyConsistentMap<FlowId, StoredFlowEntry> backupMap = backupsEnabled ? 700 + private final EventuallyConsistentMap<FlowId, StoredFlowEntry> backupMap =
643 new EventuallyConsistentMapImpl<>("flow-backup", 701 new EventuallyConsistentMapImpl<>("flow-backup",
644 clusterService, 702 clusterService,
645 clusterCommunicator, 703 clusterCommunicator,
646 flowSerializer, 704 flowSerializer,
647 clockService, 705 clockService,
648 - (key, flowEntry) -> getPeerNodes()).withTombstonesDisabled(true) 706 + (key, flowEntry) -> getPeerNodes()).withTombstonesDisabled(true);
649 - : null;
650 707
651 private Collection<NodeId> getPeerNodes() { 708 private Collection<NodeId> getPeerNodes() {
652 List<NodeId> nodes = clusterService.getNodes() 709 List<NodeId> nodes = clusterService.getNodes()
......