Support to publish port TX and RX stats in openstacktelemetry
Change-Id: I368fb676e4817cd01e5782a3b37170e2b9a5c6bd
diff --git a/apps/openstacktelemetry/app/BUCK b/apps/openstacktelemetry/app/BUCK
index 97bc0f9..4cc3d98 100644
--- a/apps/openstacktelemetry/app/BUCK
+++ b/apps/openstacktelemetry/app/BUCK
@@ -10,6 +10,7 @@
'//lib:jersey-client',
'//cli:onos-cli',
'//lib:org.apache.karaf.shell.console',
+ '//apps/openstacknode/api:onos-apps-openstacknode-api',
'//apps/openstacknetworking/api:onos-apps-openstacknetworking-api',
'//apps/openstacktelemetry/api:onos-apps-openstacktelemetry-api',
'//lib:kafka-clients',
diff --git a/apps/openstacktelemetry/app/pom.xml b/apps/openstacktelemetry/app/pom.xml
index 5b0f8bf..920958e 100644
--- a/apps/openstacktelemetry/app/pom.xml
+++ b/apps/openstacktelemetry/app/pom.xml
@@ -87,6 +87,12 @@
<dependency>
<groupId>org.onosproject</groupId>
+ <artifactId>onos-apps-openstacknode-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.onosproject</groupId>
<artifactId>onlab-osgi</artifactId>
<version>${project.version}</version>
</dependency>
diff --git a/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/codec/TinaMessageByteBufferCodec.java b/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/codec/TinaMessageByteBufferCodec.java
index 228bc2b..2ced23b 100644
--- a/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/codec/TinaMessageByteBufferCodec.java
+++ b/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/codec/TinaMessageByteBufferCodec.java
@@ -25,6 +25,9 @@
*/
public class TinaMessageByteBufferCodec {
+ public static final String KAFKA_TOPIC = "sona.flow";
+ public static final String KAFKA_KEY = "flowdata";
+
private static final int HEADER_SIZE = 8;
private static final int ENTRY_SIZE = 88;
private static final int MILLISECONDS = 1000;
diff --git a/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/KafkaTelemetryManager.java b/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/KafkaTelemetryManager.java
index d1de54a..56a0e81 100644
--- a/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/KafkaTelemetryManager.java
+++ b/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/KafkaTelemetryManager.java
@@ -124,7 +124,7 @@
public Future<RecordMetadata> publish(ProducerRecord<String, byte[]> record) {
if (producer == null) {
- log.warn("Kafka telemetry service has not been enabled!");
+ log.debug("Kafka telemetry service has not been enabled!");
return null;
}
diff --git a/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/OpenstackTelemetryManager.java b/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/OpenstackTelemetryManager.java
index da492a9..f3f386f 100644
--- a/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/OpenstackTelemetryManager.java
+++ b/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/OpenstackTelemetryManager.java
@@ -36,6 +36,9 @@
import java.util.List;
import java.util.Set;
+import static org.onosproject.openstacktelemetry.codec.TinaMessageByteBufferCodec.KAFKA_KEY;
+import static org.onosproject.openstacktelemetry.codec.TinaMessageByteBufferCodec.KAFKA_TOPIC;
+
/**
* Openstack telemetry manager.
*/
@@ -45,9 +48,6 @@
private final Logger log = LoggerFactory.getLogger(getClass());
- private static final String KAFKA_TOPIC = "sona.flow";
- private static final String KAFKA_KEY = "flowdata";
-
private List<TelemetryService> telemetryServices = Lists.newArrayList();
@Activate
@@ -89,6 +89,8 @@
if (service instanceof RestTelemetryManager) {
invokeRestPublisher((RestTelemetryService) service, flowInfos);
}
+
+ log.trace("Publishing Flow Infos {}", flowInfos);
});
}
diff --git a/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/StatsFlowRuleManager.java b/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/StatsFlowRuleManager.java
index 9025224..a7ead8e 100644
--- a/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/StatsFlowRuleManager.java
+++ b/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/StatsFlowRuleManager.java
@@ -16,7 +16,6 @@
package org.onosproject.openstacktelemetry.impl;
import com.google.common.collect.Sets;
-import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
@@ -29,12 +28,17 @@
import org.onlab.packet.IpPrefix;
import org.onlab.packet.MacAddress;
import org.onlab.packet.VlanId;
+import org.onlab.util.SharedScheduledExecutorService;
+import org.onlab.util.SharedScheduledExecutors;
import org.onosproject.cfg.ComponentConfigService;
import org.onosproject.core.ApplicationId;
import org.onosproject.core.CoreService;
import org.onosproject.mastership.MastershipService;
import org.onosproject.net.DeviceId;
import org.onosproject.net.Host;
+import org.onosproject.net.PortNumber;
+import org.onosproject.net.device.DeviceService;
+import org.onosproject.net.device.PortStatistics;
import org.onosproject.net.flow.DefaultFlowRule;
import org.onosproject.net.flow.DefaultTrafficSelector;
import org.onosproject.net.flow.DefaultTrafficTreatment;
@@ -51,7 +55,11 @@
import org.onosproject.net.flow.criteria.TcpPortCriterion;
import org.onosproject.net.flow.criteria.UdpPortCriterion;
import org.onosproject.net.host.HostService;
+import org.onosproject.openstacknetworking.api.InstancePort;
+import org.onosproject.openstacknetworking.api.InstancePortService;
import org.onosproject.openstacknetworking.api.OpenstackNetworkService;
+import org.onosproject.openstacknode.api.OpenstackNode;
+import org.onosproject.openstacknode.api.OpenstackNodeService;
import org.onosproject.openstacktelemetry.api.FlowInfo;
import org.onosproject.openstacktelemetry.api.OpenstackTelemetryService;
import org.onosproject.openstacktelemetry.api.StatsFlowRule;
@@ -62,10 +70,12 @@
import org.slf4j.LoggerFactory;
import java.util.Dictionary;
+import java.util.List;
import java.util.Optional;
import java.util.Set;
-import java.util.Timer;
-import java.util.TimerTask;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import static org.onlab.packet.Ethernet.TYPE_IPV4;
import static org.onlab.packet.IPv4.PROTOCOL_TCP;
@@ -83,6 +93,7 @@
import static org.onosproject.openstacknetworking.api.Constants.VTAP_FLAT_OUTBOUND_TABLE;
import static org.onosproject.openstacknetworking.api.Constants.VTAP_INBOUND_TABLE;
import static org.onosproject.openstacknetworking.api.Constants.VTAP_OUTBOUND_TABLE;
+import static org.onosproject.openstacknode.api.OpenstackNode.NodeType.COMPUTE;
import static org.onosproject.openstacktelemetry.api.Constants.FLAT;
import static org.onosproject.openstacktelemetry.api.Constants.OPENSTACK_TELEMETRY_APP_ID;
import static org.onosproject.openstacktelemetry.api.Constants.VLAN;
@@ -101,16 +112,28 @@
private static final byte FLOW_TYPE_SONA = 1; // VLAN
private static final long MILLISECONDS = 1000L;
+ private static final long INITIAL_DELAY = 5L;
private static final long REFRESH_INTERVAL = 5L;
+ private static final TimeUnit TIME_UNIT_SECOND = TimeUnit.SECONDS;
private static final String REVERSE_PATH_STATS = "reversePathStats";
private static final String EGRESS_STATS = "egressStats";
+ private static final String PORT_STATS = "portStats";
private static final boolean DEFAULT_REVERSE_PATH_STATS = false;
private static final boolean DEFAULT_EGRESS_STATS = false;
+ private static final boolean DEFAULT_PORT_STATS = true;
private static final String MAC_NOT_NULL = "MAC should not be null";
+ private static final String ARBITRARY_IP = "0.0.0.0/32";
+ private static final int ARBITRARY_LENGTH = 32;
+ private static final String ARBITRARY_MAC = "00:00:00:00:00:00";
+ private static final int ARBITRARY_IN_INTF = 0;
+ private static final int ARBITRARY_OUT_INTF = 0;
+
+ private static final boolean RECOVER_FROM_FAILURE = true;
+
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected CoreService coreService;
@@ -121,6 +144,9 @@
protected HostService hostService;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected DeviceService deviceService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected ComponentConfigService componentConfigService;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
@@ -130,6 +156,12 @@
protected OpenstackNetworkService osNetworkService;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected InstancePortService instPortService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected OpenstackNodeService osNodeService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected OpenstackTelemetryService telemetryService;
@Property(name = REVERSE_PATH_STATS, boolValue = DEFAULT_REVERSE_PATH_STATS,
@@ -142,12 +174,16 @@
"collecting the flow-based stats for egress port.")
private boolean egressStats = DEFAULT_EGRESS_STATS;
+ @Property(name = PORT_STATS, boolValue = DEFAULT_PORT_STATS,
+ label = "A flag which indicates whether to collect port TX & RX stats.")
+ private boolean portStats = DEFAULT_PORT_STATS;
+
private ApplicationId appId;
- private Timer timer;
- private TimerTask task;
+ private TelemetryCollector collector;
+ private SharedScheduledExecutorService executor;
+ private ScheduledFuture result;
private final Set<FlowInfo> gFlowInfoSet = Sets.newHashSet();
- private int loopCount = 0;
private static final int SOURCE_ID = 1;
private static final int TARGET_ID = 2;
@@ -157,15 +193,12 @@
private static final MacAddress NO_HOST_MAC = MacAddress.valueOf("00:00:00:00:00:00");
- public StatsFlowRuleManager() {
- this.timer = new Timer("openstack-telemetry-sender");
- }
-
@Activate
protected void activate() {
appId = coreService.registerApplication(OPENSTACK_TELEMETRY_APP_ID);
componentConfigService.registerProperties(getClass());
+ executor = SharedScheduledExecutors.getSingleThreadExecutor();
this.start();
@@ -177,6 +210,8 @@
componentConfigService.unregisterProperties(getClass(), false);
+ flowRuleService.removeFlowRulesById(appId);
+
log.info("Stopped");
}
@@ -190,16 +225,17 @@
@Override
public void start() {
log.info("Start publishing thread");
- task = new InternalTimerTask();
- timer.scheduleAtFixedRate(task, MILLISECONDS * REFRESH_INTERVAL,
- MILLISECONDS * REFRESH_INTERVAL);
+ collector = new TelemetryCollector();
+
+ result = executor.scheduleAtFixedRate(collector, INITIAL_DELAY,
+ REFRESH_INTERVAL, TIME_UNIT_SECOND, RECOVER_FROM_FAILURE);
}
@Override
public void stop() {
log.info("Stop data publishing thread");
- task.cancel();
- task = null;
+ result.cancel(true);
+ collector = null;
}
@Override
@@ -210,12 +246,221 @@
@Override
public void deleteStatFlowRule(StatsFlowRule statsFlowRule) {
- // FIXME: following code might not be necessary
- flowRuleService.removeFlowRulesById(appId);
setStatFlowRule(statsFlowRule, false);
}
+ /**
+ * Gets a set of the flow infos.
+ *
+ * @return a set of flow infos
+ */
+ @Override
+ public Set<FlowInfo> getFlowInfos() {
+ Set<FlowInfo> flowInfos = Sets.newConcurrentHashSet();
+
+ // obtain all flow rule entries installed by telemetry app
+ for (FlowEntry entry : flowRuleService.getFlowEntriesById(appId)) {
+ FlowInfo.Builder fBuilder = new DefaultFlowInfo.DefaultBuilder();
+ TrafficSelector selector = entry.selector();
+
+ IPCriterion srcIp = (IPCriterion) selector.getCriterion(IPV4_SRC);
+ IPCriterion dstIp = (IPCriterion) selector.getCriterion(IPV4_DST);
+ IPProtocolCriterion ipProtocol =
+ (IPProtocolCriterion) selector.getCriterion(IP_PROTO);
+
+ log.debug("[FlowInfo] TableID:{} SRC_IP:{} DST_IP:{} Pkt:{} Byte:{}",
+ ((IndexTableId) entry.table()).id(),
+ srcIp.ip().toString(),
+ dstIp.ip().toString(),
+ entry.packets(),
+ entry.bytes());
+
+ fBuilder.withFlowType(FLOW_TYPE_SONA)
+ .withSrcIp(srcIp.ip())
+ .withDstIp(dstIp.ip());
+
+ if (ipProtocol != null) {
+ fBuilder.withProtocol((byte) ipProtocol.protocol());
+
+ if (ipProtocol.protocol() == PROTOCOL_TCP) {
+ TcpPortCriterion tcpSrc =
+ (TcpPortCriterion) selector.getCriterion(TCP_SRC);
+ TcpPortCriterion tcpDst =
+ (TcpPortCriterion) selector.getCriterion(TCP_DST);
+
+ log.debug("TCP SRC Port: {}, DST Port: {}",
+ tcpSrc.tcpPort().toInt(),
+ tcpDst.tcpPort().toInt());
+
+ fBuilder.withSrcPort(tcpSrc.tcpPort());
+ fBuilder.withDstPort(tcpDst.tcpPort());
+
+ } else if (ipProtocol.protocol() == PROTOCOL_UDP) {
+
+ UdpPortCriterion udpSrc =
+ (UdpPortCriterion) selector.getCriterion(UDP_SRC);
+ UdpPortCriterion udpDst =
+ (UdpPortCriterion) selector.getCriterion(UDP_DST);
+
+ log.debug("UDP SRC Port: {}, DST Port: {}",
+ udpSrc.udpPort().toInt(),
+ udpDst.udpPort().toInt());
+
+ fBuilder.withSrcPort(udpSrc.udpPort());
+ fBuilder.withDstPort(udpDst.udpPort());
+ } else {
+ log.debug("Other protocol: {}", ipProtocol.protocol());
+ }
+ }
+
+ fBuilder.withSrcMac(getMacAddress(srcIp.ip().address()))
+ .withDstMac(getMacAddress(dstIp.ip().address()))
+ .withInputInterfaceId(getInterfaceId(srcIp.ip().address()))
+ .withOutputInterfaceId(getInterfaceId(dstIp.ip().address()))
+ .withVlanId(getVlanId(srcIp.ip().address()))
+ .withDeviceId(entry.deviceId());
+
+ StatsInfo.Builder sBuilder = new DefaultStatsInfo.DefaultBuilder();
+
+ // TODO: need to collect error and drop packets stats
+ // TODO: need to make the refresh interval configurable
+ sBuilder.withStartupTime(System.currentTimeMillis())
+ .withFstPktArrTime(System.currentTimeMillis())
+ .withLstPktOffset((int) (REFRESH_INTERVAL * MILLISECONDS))
+ .withCurrAccPkts((int) entry.packets())
+ .withCurrAccBytes(entry.bytes())
+ .withErrorPkts((short) 0)
+ .withDropPkts((short) 0);
+
+ fBuilder.withStatsInfo(sBuilder.build());
+
+ FlowInfo flowInfo = mergeFlowInfo(fBuilder.build(), fBuilder, sBuilder);
+
+ flowInfos.add(flowInfo);
+
+ log.debug("FlowInfo: \n{}", flowInfo.toString());
+ }
+
+ return flowInfos;
+ }
+
+ /**
+ * Gets a set of flow infos by referring to destination VM port.
+ *
+ * @return flow infos
+ */
+ private Set<FlowInfo> getDstPortBasedFlowInfos() {
+ Set<FlowInfo> flowInfos = Sets.newConcurrentHashSet();
+ Set<PortNumber> instPortNums = instPortService.instancePorts()
+ .stream()
+ .map(InstancePort::portNumber)
+ .collect(Collectors.toSet());
+ Set<DeviceId> deviceIds = osNodeService.completeNodes(COMPUTE)
+ .stream()
+ .map(OpenstackNode::intgBridge)
+ .collect(Collectors.toSet());
+
+ deviceIds.forEach(d -> {
+ List<PortStatistics> stats =
+ deviceService.getPortStatistics(d)
+ .stream()
+ .filter(s -> instPortNums.contains(s.portNumber()))
+ .collect(Collectors.toList());
+
+ stats.forEach(s -> {
+ InstancePort instPort = getInstancePort(d, s.portNumber());
+ flowInfos.add(buildTxPortInfo(instPort, s));
+ flowInfos.add(buildRxPortInfo(instPort, s));
+ });
+ });
+
+ return flowInfos;
+ }
+
+ /**
+ * Obtains the flow info generated by TX port.
+ *
+ * @param instPort instance port
+ * @param stat port statistics
+ * @return flow info
+ */
+ private FlowInfo buildTxPortInfo(InstancePort instPort, PortStatistics stat) {
+ FlowInfo.Builder fBuilder = new DefaultFlowInfo.DefaultBuilder();
+
+ fBuilder.withFlowType(FLOW_TYPE_SONA)
+ .withSrcIp(IpPrefix.valueOf(instPort.ipAddress(), ARBITRARY_LENGTH))
+ .withDstIp(IpPrefix.valueOf(ARBITRARY_IP))
+ .withSrcMac(instPort.macAddress())
+ .withDstMac(MacAddress.valueOf(ARBITRARY_MAC))
+ .withDeviceId(instPort.deviceId())
+ .withInputInterfaceId(ARBITRARY_IN_INTF)
+ .withOutputInterfaceId(ARBITRARY_OUT_INTF)
+ .withVlanId(VlanId.vlanId());
+
+ StatsInfo.Builder sBuilder = new DefaultStatsInfo.DefaultBuilder();
+ sBuilder.withStartupTime(System.currentTimeMillis())
+ .withFstPktArrTime(System.currentTimeMillis())
+ .withLstPktOffset((int) (REFRESH_INTERVAL * MILLISECONDS))
+ .withCurrAccPkts((int) stat.packetsSent())
+ .withCurrAccBytes(stat.bytesSent())
+ .withErrorPkts((short) stat.packetsTxErrors())
+ .withDropPkts((short) stat.packetsTxDropped());
+
+ fBuilder.withStatsInfo(sBuilder.build());
+
+ return mergeFlowInfo(fBuilder.build(), fBuilder, sBuilder);
+ }
+
+ /**
+ * Obtains the flow info generated by RX port.
+ *
+ * @param instPort instance port
+ * @param stat port statistics
+ * @return flow info
+ */
+ private FlowInfo buildRxPortInfo(InstancePort instPort, PortStatistics stat) {
+ FlowInfo.Builder fBuilder = new DefaultFlowInfo.DefaultBuilder();
+
+ fBuilder.withFlowType(FLOW_TYPE_SONA)
+ .withSrcIp(IpPrefix.valueOf(ARBITRARY_IP))
+ .withDstIp(IpPrefix.valueOf(instPort.ipAddress(), ARBITRARY_LENGTH))
+ .withSrcMac(MacAddress.valueOf(ARBITRARY_MAC))
+ .withDstMac(instPort.macAddress())
+ .withDeviceId(instPort.deviceId())
+ .withInputInterfaceId(ARBITRARY_IN_INTF)
+ .withOutputInterfaceId(ARBITRARY_OUT_INTF)
+ .withVlanId(VlanId.vlanId());
+
+ StatsInfo.Builder sBuilder = new DefaultStatsInfo.DefaultBuilder();
+ sBuilder.withStartupTime(System.currentTimeMillis())
+ .withFstPktArrTime(System.currentTimeMillis())
+ .withLstPktOffset((int) (REFRESH_INTERVAL * MILLISECONDS))
+ .withCurrAccPkts((int) stat.packetsReceived())
+ .withCurrAccBytes(stat.bytesReceived())
+ .withErrorPkts((short) stat.packetsRxErrors())
+ .withDropPkts((short) stat.packetsRxDropped());
+
+ fBuilder.withStatsInfo(sBuilder.build());
+
+ return mergeFlowInfo(fBuilder.build(), fBuilder, sBuilder);
+ }
+
+ /**
+ * Obtains instance port which associated with the given device identifier
+ * and port number.
+ *
+ * @param deviceId device identifier
+ * @param portNumber port number
+ * @return instance port
+ */
+ private InstancePort getInstancePort(DeviceId deviceId, PortNumber portNumber) {
+ return instPortService.instancePorts().stream()
+ .filter(p -> p.deviceId().equals(deviceId))
+ .filter(p -> p.portNumber().equals(portNumber))
+ .findFirst().orElse(null);
+ }
+
private void connectTables(DeviceId deviceId, int fromTable, int toTable,
StatsFlowRule statsFlowRule, int rulePriority,
boolean install) {
@@ -278,111 +523,19 @@
flowRuleService.apply(flowOpsBuilder.build(new FlowRuleOperationsContext() {
@Override
public void onSuccess(FlowRuleOperations ops) {
- log.debug("Provisioned vni or forwarding table: \n {}", ops.toString());
+ log.debug("Install rules for telemetry stats: \n {}",
+ ops.toString());
}
@Override
public void onError(FlowRuleOperations ops) {
- log.debug("Failed to provision vni or forwarding table: \n {}", ops.toString());
+ log.debug("Failed to install rules for telemetry stats: \n {}",
+ ops.toString());
}
}));
}
/**
- * Gets a set of the flow infos.
- *
- * @return a set of flow infos
- */
- public Set<FlowInfo> getFlowInfo() {
- Set<FlowInfo> flowInfos = Sets.newConcurrentHashSet();
-
- // obtain all flow rule entries installed by telemetry app
- for (FlowEntry entry : flowRuleService.getFlowEntriesById(appId)) {
- FlowInfo.Builder fBuilder = new DefaultFlowInfo.DefaultBuilder();
- TrafficSelector selector = entry.selector();
-
- IPCriterion srcIp = (IPCriterion) selector.getCriterion(IPV4_SRC);
- IPCriterion dstIp = (IPCriterion) selector.getCriterion(IPV4_DST);
- IPProtocolCriterion ipProtocol =
- (IPProtocolCriterion) selector.getCriterion(IP_PROTO);
-
- log.debug("[FlowInfo] TableID:{} SRC_IP:{} DST_IP:{} Pkt:{} Byte:{}",
- ((IndexTableId) entry.table()).id(),
- srcIp.ip().toString(),
- dstIp.ip().toString(),
- entry.packets(),
- entry.bytes());
-
- fBuilder.withFlowType(FLOW_TYPE_SONA)
- .withSrcIp(srcIp.ip())
- .withDstIp(dstIp.ip());
-
- if (ipProtocol != null) {
- fBuilder.withProtocol((byte) ipProtocol.protocol());
-
- if (ipProtocol.protocol() == PROTOCOL_TCP) {
- TcpPortCriterion tcpSrc =
- (TcpPortCriterion) selector.getCriterion(TCP_SRC);
- TcpPortCriterion tcpDst =
- (TcpPortCriterion) selector.getCriterion(TCP_DST);
-
- log.debug("TCP SRC Port: {}, DST Port: {}",
- tcpSrc.tcpPort().toInt(),
- tcpDst.tcpPort().toInt());
-
- fBuilder.withSrcPort(tcpSrc.tcpPort());
- fBuilder.withDstPort(tcpDst.tcpPort());
-
- } else if (ipProtocol.protocol() == PROTOCOL_UDP) {
-
- UdpPortCriterion udpSrc =
- (UdpPortCriterion) selector.getCriterion(UDP_SRC);
- UdpPortCriterion udpDst =
- (UdpPortCriterion) selector.getCriterion(UDP_DST);
-
- log.debug("UDP SRC Port: {}, DST Port: {}",
- udpSrc.udpPort().toInt(),
- udpDst.udpPort().toInt());
-
- fBuilder.withSrcPort(udpSrc.udpPort());
- fBuilder.withDstPort(udpDst.udpPort());
- } else {
- log.debug("Other protocol: {}", ipProtocol.protocol());
- }
- }
-
- fBuilder.withSrcMac(getMacAddress(srcIp.ip().address()))
- .withDstMac(getMacAddress(dstIp.ip().address()))
- .withInputInterfaceId(getInterfaceId(srcIp.ip().address()))
- .withOutputInterfaceId(getInterfaceId(dstIp.ip().address()))
- .withVlanId(getVlanId(srcIp.ip().address()))
- .withDeviceId(entry.deviceId());
-
- StatsInfo.Builder sBuilder = new DefaultStatsInfo.DefaultBuilder();
-
- // TODO: need to collect error and drop packets stats
- // TODO: need to make the refresh interval configurable
- sBuilder.withStartupTime(System.currentTimeMillis())
- .withFstPktArrTime(System.currentTimeMillis())
- .withLstPktOffset((int) (REFRESH_INTERVAL * MILLISECONDS))
- .withCurrAccPkts((int) entry.packets())
- .withCurrAccBytes(entry.bytes())
- .withErrorPkts((short) 0)
- .withDropPkts((short) 0);
-
- fBuilder.withStatsInfo(sBuilder.build());
-
- FlowInfo flowInfo = mergeFlowInfo(fBuilder.build(), fBuilder, sBuilder);
-
- flowInfos.add(flowInfo);
-
- log.debug("FlowInfo: \n{}", flowInfo.toString());
- }
-
- return flowInfos;
- }
-
- /**
* Merges old FlowInfo.StatsInfo and current FlowInfo.StatsInfo.
*
* @param flowInfo current FlowInfo object
@@ -515,9 +668,7 @@
Optional<Host> host = hostService.getHostsByIp(ipAddress).stream().findAny();
return host.map(host1 -> host1.location().deviceId()).orElse(null);
} else {
- log.warn("Failed to get DeviceID which is connected to {}. " +
- "The destination is either a bare-metal or located out of DC",
- ipAddress.toString());
+ log.debug("No DeviceID is associated to {}", ipAddress.toString());
return null;
}
}
@@ -593,29 +744,54 @@
egressStats = egressStatsConfigured;
log.info("Configured. Egress stats flag is {}", egressStats);
}
+
+ Boolean portStatsConfigured = getBooleanProperty(properties, PORT_STATS);
+ if (portStatsConfigured == null) {
+ portStats = DEFAULT_PORT_STATS;
+ log.info("Port stats flag is NOT " +
+ "configured, default value is {}", portStats);
+ } else {
+ portStats = portStatsConfigured;
+ log.info("Configured. Port stats flag is {}", portStats);
+ }
}
- private class InternalTimerTask extends TimerTask {
+ private class TelemetryCollector implements Runnable {
@Override
public void run() {
- log.debug("Timer task thread starts ({})", loopCount++);
-
Set<FlowInfo> filteredFlowInfos = Sets.newConcurrentHashSet();
// we only let the master controller of the device where the
- // stats flow rules are installed send kafka message
- getFlowInfo().forEach(f -> {
- DeviceId deviceId = getDeviceId(f.srcIp().address());
- if (mastershipService.isLocalMaster(deviceId)) {
+ // stats flow rules are installed send stats message
+ getFlowInfos().forEach(f -> {
+ if (checkSrcDstLocalMaster(f)) {
filteredFlowInfos.add(f);
}
});
- try {
- telemetryService.publish(filteredFlowInfos);
- } catch (Exception ex) {
- log.error("Exception Stack:\n{}", ExceptionUtils.getStackTrace(ex));
+ // we only let the master controller of the device where the port
+ // is located to send stats message
+ if (portStats) {
+ getDstPortBasedFlowInfos().forEach(f -> {
+ if (checkSrcDstLocalMaster(f)) {
+ filteredFlowInfos.add(f);
+ }
+ });
}
+
+ telemetryService.publish(filteredFlowInfos);
+ }
+
+ private boolean checkSrcDstLocalMaster(FlowInfo info) {
+ DeviceId srcDeviceId = getDeviceId(info.srcIp().address());
+ DeviceId dstDeviceId = getDeviceId(info.dstIp().address());
+
+ boolean isSrcLocalMaster = srcDeviceId != null &&
+ mastershipService.isLocalMaster(srcDeviceId);
+ boolean isDstLocalMaster = dstDeviceId != null &&
+ mastershipService.isLocalMaster(dstDeviceId);
+
+ return isSrcLocalMaster || isDstLocalMaster;
}
}
}
diff --git a/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/web/OpenstackTelemetryWebResource.java b/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/web/OpenstackTelemetryWebResource.java
index 2efab8b..419b789 100644
--- a/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/web/OpenstackTelemetryWebResource.java
+++ b/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/web/OpenstackTelemetryWebResource.java
@@ -30,6 +30,7 @@
import org.slf4j.LoggerFactory;
import javax.ws.rs.Consumes;
+import javax.ws.rs.DELETE;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
@@ -98,6 +99,9 @@
* @param input openstack flow rule JSON input stream
* @return 200 OK if processing is correct.
*/
+ @DELETE
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Produces(MediaType.APPLICATION_JSON)
public Response deleteBulkFlowRule(InputStream input) {
log.info("DELETE BULK FLOW RULE: {}", input.toString());
@@ -132,7 +136,7 @@
log.info("GET BULK FLOW RULE");
Set<FlowInfo> flowInfoSet;
- flowInfoSet = statsFlowRuleService.getFlowInfo();
+ flowInfoSet = statsFlowRuleService.getFlowInfos();
log.info("\n\n======================================================\n" +
"FlowInfo Set: \n{}" +
@@ -151,11 +155,10 @@
}
@GET
- @Path("list/{src_ip_prefix}/{dst_ip_prefix}")
+ @Path("list/{srcIpPrefix}/{dstIpPrefix}")
@Produces(MediaType.APPLICATION_JSON)
- public Response getFlowRule(
- @PathParam("src_ip_prefix") String srcIpPrefix,
- @PathParam("dst_ip_prefix") String dstIpPrefix) {
+ public Response getFlowRule(@PathParam("srcIpPrefix") String srcIpPrefix,
+ @PathParam("dstIpPrefix") String dstIpPrefix) {
return ok(root).build();
}