+++ /dev/null
-/**
- * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-
-package org.opendaylight.controller.md.statistics.manager.impl;
-
-import java.util.Arrays;
-import java.util.List;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
-
-import org.opendaylight.controller.md.statistics.manager.StatRpcMsgManager;
-import org.opendaylight.controller.md.statistics.manager.StatisticsManager;
-import org.opendaylight.controller.sal.binding.api.RpcConsumerRegistry;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetAllFlowsStatisticsFromAllFlowTablesInputBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.OpendaylightFlowStatisticsService;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.GetFlowTablesStatisticsInputBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.table.statistics.rev131215.OpendaylightFlowTableStatisticsService;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev131103.TransactionAware;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev131103.TransactionId;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GetAllGroupStatisticsInputBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GetGroupDescriptionInputBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.GetGroupFeaturesInputBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.group.statistics.rev131111.OpendaylightGroupStatisticsService;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.GetAllMeterConfigStatisticsInputBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.GetAllMeterStatisticsInputBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.GetMeterFeaturesInputBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.statistics.rev131111.OpendaylightMeterStatisticsService;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.GetAllNodeConnectorsStatisticsInputBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.GetAllNodeConnectorsStatisticsOutput;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.OpendaylightPortStatisticsService;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.GetAllQueuesStatisticsFromAllPortsInputBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.queue.statistics.rev131216.OpendaylightQueueStatisticsService;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.TableId;
-import org.opendaylight.yangtools.yang.binding.DataObject;
-import org.opendaylight.yangtools.yang.common.RpcResult;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Joiner;
-import com.google.common.base.Optional;
-import com.google.common.base.Preconditions;
-import com.google.common.cache.Cache;
-import com.google.common.cache.CacheBuilder;
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.JdkFutureAdapters;
-import com.google.common.util.concurrent.SettableFuture;
-
-
-/**
- * statistics-manager
- * org.opendaylight.controller.md.statistics.manager.impl
- *
- * StatRpcMsgManagerImpl
- * Class register and provide all RPC Statistics Device Services and implement pre-defined
- * wrapped methods for prepare easy access to RPC Statistics Device Services like getAllStatisticsFor...
- *
- * In next Class implement process for joining multipart messages.
- * Class internally use two WeakHashMap and GuavaCache for holding values for joining multipart msg.
- * One Weak map is used for holding all Multipart Messages and second is used for possible input
- * Config/DS light-weight DataObject (DataObject contains only necessary identification fields as
- * TableId, GroupId, MeterId or for flow Match, Priority, FlowCookie, TableId and FlowId ...
- *
- * @author avishnoi@in.ibm.com <a href="mailto:vdemcak@cisco.com">Vaclav Demcak</a>
- *
- */
-public class StatRpcMsgManagerImpl implements StatRpcMsgManager {
-
- private final static Logger LOG = LoggerFactory.getLogger(StatRpcMsgManagerImpl.class);
-
- private final Cache<String, TransactionCacheContainer<? super TransactionAware>> txCache;
-
- private final int queueCapacity = 5000;
-
- private final OpendaylightGroupStatisticsService groupStatsService;
- private final OpendaylightMeterStatisticsService meterStatsService;
- private final OpendaylightFlowStatisticsService flowStatsService;
- private final OpendaylightPortStatisticsService portStatsService;
- private final OpendaylightFlowTableStatisticsService flowTableStatsService;
- private final OpendaylightQueueStatisticsService queueStatsService;
-
- private BlockingQueue<RpcJobsQueue> statsRpcJobQueue;
-
- private volatile boolean finishing = false;
-
- public StatRpcMsgManagerImpl (final StatisticsManager manager,
- final RpcConsumerRegistry rpcRegistry, final long maxNodeForCollector) {
- Preconditions.checkArgument(manager != null, "StatisticManager can not be null!");
- Preconditions.checkArgument(rpcRegistry != null, "RpcConsumerRegistry can not be null !");
- groupStatsService = Preconditions.checkNotNull(
- rpcRegistry.getRpcService(OpendaylightGroupStatisticsService.class),
- "OpendaylightGroupStatisticsService can not be null!");
- meterStatsService = Preconditions.checkNotNull(
- rpcRegistry.getRpcService(OpendaylightMeterStatisticsService.class),
- "OpendaylightMeterStatisticsService can not be null!");
- flowStatsService = Preconditions.checkNotNull(
- rpcRegistry.getRpcService(OpendaylightFlowStatisticsService.class),
- "OpendaylightFlowStatisticsService can not be null!");
- portStatsService = Preconditions.checkNotNull(
- rpcRegistry.getRpcService(OpendaylightPortStatisticsService.class),
- "OpendaylightPortStatisticsService can not be null!");
- flowTableStatsService = Preconditions.checkNotNull(
- rpcRegistry.getRpcService(OpendaylightFlowTableStatisticsService.class),
- "OpendaylightFlowTableStatisticsService can not be null!");
- queueStatsService = Preconditions.checkNotNull(
- rpcRegistry.getRpcService(OpendaylightQueueStatisticsService.class),
- "OpendaylightQueueStatisticsService can not be null!");
-
- statsRpcJobQueue = new LinkedBlockingQueue<>(queueCapacity);
- /* nr. 7 is here nr. of possible statistic which are waiting for notification
- * - check it in StatPermCollectorImpl method collectStatCrossNetwork */
- txCache = CacheBuilder.newBuilder().expireAfterWrite((maxNodeForCollector * 7), TimeUnit.SECONDS)
- .maximumSize(10000).build();
- }
-
- @Override
- public void close() {
- finishing = true;
- statsRpcJobQueue = null;
- }
-
- @Override
- public void run() {
- /* Neverending cyle - wait for finishing */
- while ( ! finishing) {
- try {
- statsRpcJobQueue.take().call();
- }
- catch (final Exception e) {
- LOG.warn("Stat Element RPC executor fail!", e);
- }
- }
- // Drain all rpcCall, making sure any blocked threads are unblocked
- while ( ! statsRpcJobQueue.isEmpty()) {
- statsRpcJobQueue.poll();
- }
- }
-
- private void addGetAllStatJob(final RpcJobsQueue getAllStatJob) {
- final boolean success = statsRpcJobQueue.offer(getAllStatJob);
- if ( ! success) {
- LOG.warn("Put RPC request getAllStat fail! Queue is full.");
- }
- }
-
- private void addStatJob(final RpcJobsQueue getStatJob) {
- final boolean success = statsRpcJobQueue.offer(getStatJob);
- if ( ! success) {
- LOG.debug("Put RPC request for getStat fail! Queue is full.");
- }
- }
-
- @Override
- public <T extends TransactionAware, D extends DataObject> void registrationRpcFutureCallBack(
- final Future<RpcResult<T>> future, final D inputObj, final NodeRef nodeRef,
- final SettableFuture<TransactionId> resultTransId) {
-
- Futures.addCallback(JdkFutureAdapters.listenInPoolThread(future),
- new FutureCallback<RpcResult<? extends TransactionAware>>() {
-
- @Override
- public void onSuccess(final RpcResult<? extends TransactionAware> result) {
- final TransactionId id = result.getResult().getTransactionId();
- final NodeKey nodeKey = nodeRef.getValue().firstKeyOf(Node.class, NodeKey.class);
- if (id == null) {
- String[] multipartRequestName = result.getResult().getClass().getSimpleName().split("(?=\\p{Upper})");
- LOG.warn("Node [{}] does not support statistics request type : {}",
- nodeKey.getId(),Joiner.on(" ").join(Arrays.copyOfRange(multipartRequestName, 2, multipartRequestName.length-2)));
- } else {
- if (resultTransId != null) {
- resultTransId.set(id);
- }
- final String cacheKey = buildCacheKey(id, nodeKey.getId());
- final TransactionCacheContainer<? super TransactionAware> container =
- new TransactionCacheContainerImpl<>(id, inputObj, nodeKey.getId());
- txCache.put(cacheKey, container);
- }
- }
-
- @Override
- public void onFailure(final Throwable t) {
- LOG.warn("Response Registration for Statistics RPC call fail!", t);
- }
-
- });
- }
-
- private String buildCacheKey(final TransactionId id, final NodeId nodeId) {
- return String.valueOf(id.getValue()) + "-" + nodeId.getValue();
- }
-
- @Override
- public Future<Optional<TransactionCacheContainer<?>>> getTransactionCacheContainer(
- final TransactionId id, final NodeId nodeId) {
- Preconditions.checkArgument(id != null, "TransactionId can not be null!");
- Preconditions.checkArgument(nodeId != null, "NodeId can not be null!");
-
- final String key = buildCacheKey(id, nodeId);
- final SettableFuture<Optional<TransactionCacheContainer<?>>> result = SettableFuture.create();
-
- final RpcJobsQueue getTransactionCacheContainer = new RpcJobsQueue() {
-
- @Override
- public Void call() throws Exception {
- final Optional<TransactionCacheContainer<?>> resultContainer =
- Optional.<TransactionCacheContainer<?>> fromNullable(txCache.getIfPresent(key));
- if (resultContainer.isPresent()) {
- txCache.invalidate(key);
- }
- result.set(resultContainer);
- return null;
- }
- };
- addStatJob(getTransactionCacheContainer);
- return result;
- }
-
- @Override
- public Future<Boolean> isExpectedStatistics(final TransactionId id, final NodeId nodeId) {
- Preconditions.checkArgument(id != null, "TransactionId can not be null!");
- Preconditions.checkArgument(nodeId != null, "NodeId can not be null!");
-
- final String key = buildCacheKey(id, nodeId);
- final SettableFuture<Boolean> checkStatId = SettableFuture.create();
-
- final RpcJobsQueue isExpecedStatistics = new RpcJobsQueue() {
-
- @Override
- public Void call() throws Exception {
- final Optional<TransactionCacheContainer<?>> result =
- Optional.<TransactionCacheContainer<?>> fromNullable(txCache.getIfPresent(key));
- checkStatId.set(Boolean.valueOf(result.isPresent()));
- return null;
- }
- };
- addStatJob(isExpecedStatistics);
- return checkStatId;
- }
-
- @Override
- public void addNotification(final TransactionAware notification, final NodeId nodeId) {
- Preconditions.checkArgument(notification != null, "TransactionAware can not be null!");
- Preconditions.checkArgument(nodeId != null, "NodeId can not be null!");
-
- final RpcJobsQueue addNotification = new RpcJobsQueue() {
-
- @Override
- public Void call() throws Exception {
- final TransactionId txId = notification.getTransactionId();
- final String key = buildCacheKey(txId, nodeId);
- final TransactionCacheContainer<? super TransactionAware> container = (txCache.getIfPresent(key));
- if (container != null) {
- container.addNotif(notification);
- }
- return null;
- }
- };
- addStatJob(addNotification);
- }
-
- @Override
- public Future<TransactionId> getAllGroupsStat(final NodeRef nodeRef) {
- Preconditions.checkArgument(nodeRef != null, "NodeRef can not be null!");
- final SettableFuture<TransactionId> result = SettableFuture.create();
- final RpcJobsQueue getAllGroupStat = new RpcJobsQueue() {
-
- @Override
- public Void call() throws Exception {
- final GetAllGroupStatisticsInputBuilder builder =
- new GetAllGroupStatisticsInputBuilder();
- builder.setNode(nodeRef);
- registrationRpcFutureCallBack(groupStatsService
- .getAllGroupStatistics(builder.build()), null, nodeRef, result);
- return null;
- }
- };
- addGetAllStatJob(getAllGroupStat);
- return result;
- }
-
- @Override
- public Future<TransactionId> getAllMetersStat(final NodeRef nodeRef) {
- Preconditions.checkArgument(nodeRef != null, "NodeRef can not be null!");
- final SettableFuture<TransactionId> result = SettableFuture.create();
- final RpcJobsQueue getAllMeterStat = new RpcJobsQueue() {
-
- @Override
- public Void call() throws Exception {
- final GetAllMeterStatisticsInputBuilder builder =
- new GetAllMeterStatisticsInputBuilder();
- builder.setNode(nodeRef);
- registrationRpcFutureCallBack(meterStatsService
- .getAllMeterStatistics(builder.build()), null, nodeRef, result);
- return null;
- }
- };
- addGetAllStatJob(getAllMeterStat);
- return result;
- }
-
- @Override
- public Future<TransactionId> getAllFlowsStat(final NodeRef nodeRef) {
- Preconditions.checkArgument(nodeRef != null, "NodeRef can not be null!");
- final SettableFuture<TransactionId> result = SettableFuture.create();
- final RpcJobsQueue getAllFlowStat = new RpcJobsQueue() {
-
- @Override
- public Void call() throws Exception {
- final GetAllFlowsStatisticsFromAllFlowTablesInputBuilder builder =
- new GetAllFlowsStatisticsFromAllFlowTablesInputBuilder();
- builder.setNode(nodeRef);
- registrationRpcFutureCallBack(flowStatsService
- .getAllFlowsStatisticsFromAllFlowTables(builder.build()), null, nodeRef, result);
- return null;
- }
- };
- addGetAllStatJob(getAllFlowStat);
- return result;
- }
-
- @Override
- public void getAggregateFlowStat(final NodeRef nodeRef, final TableId tableId) {
- Preconditions.checkArgument(nodeRef != null, "NodeRef can not be null!");
- Preconditions.checkArgument(tableId != null, "TableId can not be null!");
- final RpcJobsQueue getAggregateFlowStat = new RpcJobsQueue() {
-
- @Override
- public Void call() throws Exception {
- final GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder builder =
- new GetAggregateFlowStatisticsFromFlowTableForAllFlowsInputBuilder();
- builder.setNode(nodeRef);
- builder.setTableId(tableId);
-
- final TableBuilder tbuilder = new TableBuilder();
- tbuilder.setId(tableId.getValue());
- tbuilder.setKey(new TableKey(tableId.getValue()));
- registrationRpcFutureCallBack(flowStatsService
- .getAggregateFlowStatisticsFromFlowTableForAllFlows(builder.build()), tbuilder.build(), nodeRef, null);
- return null;
- }
- };
- addGetAllStatJob(getAggregateFlowStat);
- }
-
- @Override
- public Future<TransactionId> getAllPortsStat(final NodeRef nodeRef) {
- Preconditions.checkArgument(nodeRef != null, "NodeRef can not be null!");
- final SettableFuture<TransactionId> result = SettableFuture.create();
- final RpcJobsQueue getAllPortsStat = new RpcJobsQueue() {
-
- @Override
- public Void call() throws Exception {
- final GetAllNodeConnectorsStatisticsInputBuilder builder =
- new GetAllNodeConnectorsStatisticsInputBuilder();
- builder.setNode(nodeRef);
- final Future<RpcResult<GetAllNodeConnectorsStatisticsOutput>> rpc =
- portStatsService.getAllNodeConnectorsStatistics(builder.build());
- registrationRpcFutureCallBack(rpc, null, nodeRef, result);
- return null;
- }
- };
- addGetAllStatJob(getAllPortsStat);
- return result;
- }
-
- @Override
- public Future<TransactionId> getAllTablesStat(final NodeRef nodeRef) {
- Preconditions.checkArgument(nodeRef != null, "NodeRef can not be null!");
- final SettableFuture<TransactionId> result = SettableFuture.create();
- final RpcJobsQueue getAllTableStat = new RpcJobsQueue() {
-
- @Override
- public Void call() throws Exception {
- final GetFlowTablesStatisticsInputBuilder builder =
- new GetFlowTablesStatisticsInputBuilder();
- builder.setNode(nodeRef);
- registrationRpcFutureCallBack(flowTableStatsService
- .getFlowTablesStatistics(builder.build()), null, nodeRef, result);
- return null;
- }
- };
- addGetAllStatJob(getAllTableStat);
- return result;
- }
-
- @Override
- public Future<TransactionId> getAllQueueStat(final NodeRef nodeRef) {
- Preconditions.checkArgument(nodeRef != null, "NodeRef can not be null!");
- final SettableFuture<TransactionId> result = SettableFuture.create();
- final RpcJobsQueue getAllQueueStat = new RpcJobsQueue() {
-
- @Override
- public Void call() throws Exception {
- final GetAllQueuesStatisticsFromAllPortsInputBuilder builder =
- new GetAllQueuesStatisticsFromAllPortsInputBuilder();
- builder.setNode(nodeRef);
- registrationRpcFutureCallBack(queueStatsService
- .getAllQueuesStatisticsFromAllPorts(builder.build()), null, nodeRef, result);
- return null;
- }
- };
- addGetAllStatJob(getAllQueueStat);
- return result;
- }
-
- @Override
- public Future<TransactionId> getAllMeterConfigStat(final NodeRef nodeRef) {
- Preconditions.checkArgument(nodeRef != null, "NodeRef can not be null!");
- final SettableFuture<TransactionId> result = SettableFuture.create();
- final RpcJobsQueue qetAllMeterConfStat = new RpcJobsQueue() {
-
- @Override
- public Void call() throws Exception {
- final GetAllMeterConfigStatisticsInputBuilder builder =
- new GetAllMeterConfigStatisticsInputBuilder();
- builder.setNode(nodeRef);
- registrationRpcFutureCallBack(meterStatsService
- .getAllMeterConfigStatistics(builder.build()), null, nodeRef, result);
- return null;
- }
- };
- addGetAllStatJob(qetAllMeterConfStat);
- return result;
- }
-
- @Override
- public void getGroupFeaturesStat(final NodeRef nodeRef) {
- Preconditions.checkArgument(nodeRef != null, "NodeRef can not be null!");
- final RpcJobsQueue getGroupFeaturesStat = new RpcJobsQueue() {
-
- @Override
- public Void call() throws Exception {
- /* RPC input */
- final GetGroupFeaturesInputBuilder input = new GetGroupFeaturesInputBuilder();
- input.setNode(nodeRef);
- registrationRpcFutureCallBack(groupStatsService.getGroupFeatures(input.build()), null, nodeRef, null);
- return null;
- }
- };
- addStatJob(getGroupFeaturesStat);
- }
-
- @Override
- public void getMeterFeaturesStat(final NodeRef nodeRef) {
- Preconditions.checkArgument(nodeRef != null, "NodeRef can not be null!");
- final RpcJobsQueue getMeterFeaturesStat = new RpcJobsQueue() {
-
- @Override
- public Void call() throws Exception {
- /* RPC input */
- final GetMeterFeaturesInputBuilder input = new GetMeterFeaturesInputBuilder();
- input.setNode(nodeRef);
- registrationRpcFutureCallBack(meterStatsService.getMeterFeatures(input.build()), null, nodeRef, null);
- return null;
- }
- };
- addStatJob(getMeterFeaturesStat);
- }
-
- @Override
- public Future<TransactionId> getAllGroupsConfStats(final NodeRef nodeRef) {
- Preconditions.checkArgument(nodeRef != null, "NodeRef can not be null!");
- final SettableFuture<TransactionId> result = SettableFuture.create();
- final RpcJobsQueue getAllGropConfStat = new RpcJobsQueue() {
-
- @Override
- public Void call() throws Exception {
- final GetGroupDescriptionInputBuilder builder =
- new GetGroupDescriptionInputBuilder();
- builder.setNode(nodeRef);
- registrationRpcFutureCallBack(groupStatsService
- .getGroupDescription(builder.build()), null, nodeRef, result);
-
- return null;
- }
- };
- addGetAllStatJob(getAllGropConfStat);
- return result;
- }
-
- public class TransactionCacheContainerImpl<T extends TransactionAware> implements TransactionCacheContainer<T> {
-
- private final TransactionId id;
- private final NodeId nId;
- private final List<T> notifications;
- private final Optional<? extends DataObject> confInput;
-
- public <D extends DataObject> TransactionCacheContainerImpl (final TransactionId id, final D input, final NodeId nodeId) {
- this.id = Preconditions.checkNotNull(id, "TransactionId can not be null!");
- notifications = new CopyOnWriteArrayList<T>();
- confInput = Optional.fromNullable(input);
- nId = nodeId;
- }
-
- @Override
- public void addNotif(final T notif) {
- notifications.add(notif);
- }
-
- @Override
- public TransactionId getId() {
- return id;
- }
-
- @Override
- public NodeId getNodeId() {
- return nId;
- }
-
- @Override
- public List<T> getNotifications() {
- return notifications;
- }
-
- @Override
- public Optional<? extends DataObject> getConfInput() {
- return confInput;
- }
- }
-}
-
Code Review / controller.git / blobdiff