Skip to content

Commit

Permalink
Allows to read ingress and egress counter separately (#520)
Browse files Browse the repository at this point in the history
  • Loading branch information
@daniele-moro
daniele-moro authored May 19, 2022
1 parent 5376d8c commit fd3c3f0
Show file tree
Hide file tree
Showing 9 changed files with 651 additions and 119 deletions.
163 changes: 111 additions & 52 deletions src/main/java/org/stratumproject/fabric/tna/behaviour/upf/FabricUpfProgrammable.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import org.apache.commons.lang3.tuple.Pair;
import org.onlab.packet.Ip4Prefix;
import org.onosproject.core.ApplicationId;
Expand Down Expand Up @@ -65,8 +66,10 @@
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.LongStream;

import static java.lang.String.format;
import static org.onosproject.net.behaviour.upf.UpfProgrammableException.Type.UNSUPPORTED_OPERATION;
import static org.onosproject.net.pi.model.PiCounterType.INDIRECT;
import static org.stratumproject.fabric.tna.behaviour.P4InfoConstants.FABRIC_EGRESS_UPF_EG_TUNNEL_PEERS;
import static org.stratumproject.fabric.tna.behaviour.P4InfoConstants.FABRIC_EGRESS_UPF_GTPU_ENCAP;
Expand Down Expand Up @@ -101,6 +104,9 @@ public class FabricUpfProgrammable extends AbstractP4RuntimeHandlerBehaviour
private static final int DEFAULT_PRIORITY = 128;
private static final long DEFAULT_P4_DEVICE_ID = 1;

private static final ImmutableSet<UpfEntityType> COUNTER_TYPES =
ImmutableSet.of(UpfEntityType.COUNTER, UpfEntityType.INGRESS_COUNTER, UpfEntityType.EGRESS_COUNTER);

protected FlowRuleService flowRuleService;
protected MeterService meterService;
protected PacketService packetService;
Expand All @@ -112,7 +118,8 @@ public class FabricUpfProgrammable extends AbstractP4RuntimeHandlerBehaviour
private long downlinkUeSessionsTableSize;
private long uplinkUpfTerminationsTableSize;
private long downlinkUpfTerminationsTableSize;
private long upfCounterSize;
private long ingressUpfCounterSize;
private long egressUpfCounterSize;
private long gtpTunnelPeersTableSize;
private long applicationsTableSize;
private long appMeterSize;
Expand Down Expand Up @@ -259,7 +266,7 @@ private boolean computeHardwareResourceSizes() {
long sessionMeterSize = 0;
long appMeterSize = 0;
long sliceMeterSize = 0;
for (PiMeterModel piMeter: pipeconf.pipelineModel().meters()) {
for (PiMeterModel piMeter : pipeconf.pipelineModel().meters()) {
if (piMeter.id().equals(FABRIC_INGRESS_UPF_SESSION_METER)) {
sessionMeterSize = piMeter.size();
} else if (piMeter.id().equals(FABRIC_INGRESS_UPF_APP_METER)) {
Expand All @@ -280,7 +287,8 @@ private boolean computeHardwareResourceSizes() {
this.uplinkUpfTerminationsTableSize = uplinkUpfTerminationsTableSize;
this.downlinkUpfTerminationsTableSize = downlinkUpfTerminationsTableSize;
this.applicationsTableSize = applicationsTableSize;
this.upfCounterSize = Math.min(ingressCounterSize, egressCounterSize);
this.ingressUpfCounterSize = ingressCounterSize;
this.egressUpfCounterSize = egressCounterSize;
this.gtpTunnelPeersTableSize = Math.min(ingressGtpTunnelPeersTableSize, egressGtpTunnelPeersTableSize);
this.sessionMeterSize = sessionMeterSize;
this.sliceMeterSize = sliceMeterSize;
Expand Down Expand Up @@ -432,7 +440,11 @@ public Collection<? extends UpfEntity> readAll(UpfEntityType entityType)
case TUNNEL_PEER:
return getGtpTunnelPeers();
case COUNTER:
return readCounters(-1);
return readCounters(-1, UpfEntityType.COUNTER);
case INGRESS_COUNTER:
return readCounters(-1, UpfEntityType.INGRESS_COUNTER);
case EGRESS_COUNTER:
return readCounters(-1, UpfEntityType.EGRESS_COUNTER);
case APPLICATION:
return getUpfApplication();
case SESSION_METER:
Expand Down Expand Up @@ -548,27 +560,25 @@ private Collection<UpfEntity> getUpfTerminationsDownlink() throws UpfProgrammabl
}

@Override
public Collection<UpfCounter> readCounters(long maxCounterId) {
public Collection<UpfCounter> readCounters(long maxCounterId, UpfEntityType type) throws UpfProgrammableException {
assertCounterType(type);
if (!setupBehaviour("readCounters()")) {
return null;
}

long counterSize = upfCounterSize;
long counterSize = getEntitySize(type);
if (maxCounterId != -1) {
counterSize = Math.min(maxCounterId, counterSize);
}

// Prepare UpfCounter object builders, one for each counter ID currently in use
Map<Integer, UpfCounter.Builder> upfCounterBuilders = Maps.newHashMap();
for (int cellId = 0; cellId < counterSize; cellId++) {
upfCounterBuilders.put(cellId, UpfCounter.builder().withCellId(cellId));
}

// Generate the counter cell IDs.
Set<PiCounterId> counterIds = ImmutableSet.of(
FABRIC_INGRESS_UPF_TERMINATIONS_COUNTER,
FABRIC_EGRESS_UPF_TERMINATIONS_COUNTER
);
Set<PiCounterId> counterIds = Sets.newHashSet();
if (isIngressCounter(type) || isBiCounter(type)) {
counterIds.add(FABRIC_INGRESS_UPF_TERMINATIONS_COUNTER);
}
if (isEgressCounter(type) || isBiCounter(type)) {
counterIds.add(FABRIC_EGRESS_UPF_TERMINATIONS_COUNTER);
}

// Query the device.
Collection<PiCounterCell> counterEntryResponse = client.read(
Expand All @@ -577,20 +587,42 @@ public Collection<UpfCounter> readCounters(long maxCounterId) {
.submitSync()
.all(PiCounterCell.class);

// Process response.
counterEntryResponse.forEach(counterCell -> {
// Generate list of requested cell indexes
List<Long> cellIds = LongStream.range(0, counterSize)
.boxed().collect(Collectors.toList());

return processCounterEntryResponse(
counterEntryResponse, cellIds, type);
}

/**
* Process the counters read from the device given the expected counter indexes
* and the UpfCounter Type.
*/
private List<UpfCounter> processCounterEntryResponse(
Collection<PiCounterCell> readCounterEntries,
List<Long> expectedCounterIndex,
UpfEntityType type
) {
// Prepare UpfCounter object builders, one for each expected counter index
Map<Long, UpfCounter.Builder> upfCounterBuilders = Maps.newHashMap();
expectedCounterIndex.forEach(cellId ->
upfCounterBuilders.put(cellId, UpfCounter.builder().withCellId(cellId.intValue()))
);
readCounterEntries.forEach(counterCell -> {
if (counterCell.cellId().counterType() != INDIRECT) {
log.warn("Invalid counter data type {}, skipping", counterCell.cellId().counterType());
return;
}
if (!upfCounterBuilders.containsKey((int) counterCell.cellId().index())) {
// Most likely Up4config.maxUes() is set to a value smaller than what the switch
// pipeline can hold.
log.debug("Unrecognized index {} when reading all counters, " +
if (!expectedCounterIndex.contains(counterCell.cellId().index())) {
// This can happen if the Up4Config.maxUes() is set to a value
// smaller than what the switch pipeline can hold, and we do a
// wild card read
log.debug("Unrecognized index {} when reading counters, " +
"that's expected if we are manually limiting maxUes", counterCell);
return;
}
UpfCounter.Builder statsBuilder = upfCounterBuilders.get((int) counterCell.cellId().index());
UpfCounter.Builder statsBuilder = upfCounterBuilders.get(counterCell.cellId().index());
if (counterCell.cellId().counterId().equals(FABRIC_INGRESS_UPF_TERMINATIONS_COUNTER)) {
statsBuilder.setIngress(counterCell.data().packets(),
counterCell.data().bytes());
Expand All @@ -600,8 +632,14 @@ public Collection<UpfCounter> readCounters(long maxCounterId) {
} else {
log.warn("Unrecognized counter ID {}, skipping", counterCell);
}
// UpfCounter builder defaults to type COUNTER
if (isIngressCounter(type)) {
statsBuilder.isIngressCounter();
}
if (isEgressCounter(type)) {
statsBuilder.isEgressCounter();
}
});

return upfCounterBuilders
.values()
.stream()
Expand All @@ -614,7 +652,10 @@ public long tableSize(UpfEntityType entityType) throws UpfProgrammableException
if (!setupBehaviour("tableSize()")) {
return -1;
}
return getEntitySize(entityType);
}

private long getEntitySize(UpfEntityType entityType) throws UpfProgrammableException {
switch (entityType) {
case INTERFACE:
return interfaceTableSize;
Expand All @@ -629,7 +670,11 @@ public long tableSize(UpfEntityType entityType) throws UpfProgrammableException
case TERMINATION_DOWNLINK:
return this.downlinkUpfTerminationsTableSize;
case COUNTER:
return upfCounterSize;
return Math.min(ingressUpfCounterSize, egressUpfCounterSize);
case INGRESS_COUNTER:
return ingressUpfCounterSize;
case EGRESS_COUNTER:
return egressUpfCounterSize;
case APPLICATION:
return applicationsTableSize;
case APPLICATION_METER:
Expand All @@ -645,48 +690,38 @@ public long tableSize(UpfEntityType entityType) throws UpfProgrammableException
}

@Override
public UpfCounter readCounter(int cellId) throws UpfProgrammableException {
public UpfCounter readCounter(int cellId, UpfEntityType type) throws UpfProgrammableException {
assertCounterType(type);
if (!setupBehaviour("readCounter()")) {
return null;
}
if (cellId >= upfCounterSize || cellId < 0) {
if (cellId >= getEntitySize(type) || cellId < 0) {
throw new UpfProgrammableException("Requested UPF counter cell index is out of bounds.",
UpfProgrammableException.Type.ENTITY_OUT_OF_RANGE);
}
UpfCounter.Builder stats = UpfCounter.builder().withCellId(cellId);

// Make list of cell handles we want to read.
List<PiCounterCellHandle> counterCellHandles = List.of(
PiCounterCellHandle.of(deviceId,
PiCounterCellId.ofIndirect(FABRIC_INGRESS_UPF_TERMINATIONS_COUNTER, cellId)),
PiCounterCellHandle.of(deviceId,
PiCounterCellId.ofIndirect(FABRIC_EGRESS_UPF_TERMINATIONS_COUNTER, cellId)));
List<PiCounterCellHandle> counterCellHandles = Lists.newArrayList();
if (isIngressCounter(type) || isBiCounter(type)) {
counterCellHandles.add(PiCounterCellHandle.of(
deviceId, PiCounterCellId.ofIndirect(FABRIC_INGRESS_UPF_TERMINATIONS_COUNTER, cellId)
));
}
if (isEgressCounter(type) || isBiCounter(type)) {
counterCellHandles.add(PiCounterCellHandle.of(
deviceId, PiCounterCellId.ofIndirect(FABRIC_EGRESS_UPF_TERMINATIONS_COUNTER, cellId)
));
}

// Query the device.
Collection<PiCounterCell> counterEntryResponse = client.read(
DEFAULT_P4_DEVICE_ID, pipeconf)
.handles(counterCellHandles).submitSync()
.all(PiCounterCell.class);

// Process response.
counterEntryResponse.forEach(counterCell -> {
if (counterCell.cellId().counterType() != INDIRECT) {
log.warn("Invalid counter data type {}, skipping", counterCell.cellId().counterType());
return;
}
if (cellId != counterCell.cellId().index()) {
log.warn("Unrecognized counter index {}, skipping", counterCell);
return;
}
if (counterCell.cellId().counterId().equals(FABRIC_INGRESS_UPF_TERMINATIONS_COUNTER)) {
stats.setIngress(counterCell.data().packets(), counterCell.data().bytes());
} else if (counterCell.cellId().counterId().equals(FABRIC_EGRESS_UPF_TERMINATIONS_COUNTER)) {
stats.setEgress(counterCell.data().packets(), counterCell.data().bytes());
} else {
log.warn("Unrecognized counter ID {}, skipping", counterCell);
}
});
return stats.build();
return processCounterEntryResponse(counterEntryResponse, Lists.newArrayList((long) cellId), type)
.get(0);
}

@Override
Expand Down Expand Up @@ -723,6 +758,8 @@ public void apply(UpfEntity entity) throws UpfProgrammableException {
applyUpfMeter((UpfMeter) entity);
break;
case COUNTER:
case INGRESS_COUNTER:
case EGRESS_COUNTER:
default:
throw new UpfProgrammableException(format("Adding entity type %s not supported.",
entity.type().humanReadableName()));
Expand Down Expand Up @@ -854,8 +891,8 @@ public void delete(UpfEntity entity) throws UpfProgrammableException {
case SESSION_METER:
case APPLICATION_METER:
case SLICE_METER:
// Meter cannot be deleted, only modified.
case COUNTER:
// Entities cannot be deleted, only modified.
default:
throw new UpfProgrammableException(format("Deleting entity type %s not supported.",
entity.type().humanReadableName()));
Expand All @@ -866,6 +903,7 @@ private boolean removeEntry(PiCriterion match, PiTableId tableId, boolean failSi
throws UpfProgrammableException {
return removeEntry(match, tableId, failSilent, DEFAULT_PRIORITY);
}

private boolean removeEntry(PiCriterion match, PiTableId tableId, boolean failSilent, int priority)
throws UpfProgrammableException {
return removeEntries(Lists.newArrayList(Pair.of(tableId, match)), failSilent, priority);
Expand Down Expand Up @@ -1034,4 +1072,25 @@ private void assertTrafficClass(int sliceId, int tc) throws UpfProgrammableExcep
));
}
}

private void assertCounterType(UpfEntityType type) throws UpfProgrammableException {
if (!COUNTER_TYPES.contains(type)) {
throw new UpfProgrammableException(
String.format("Unsupported counter type (%s)!", type.toString()),
UNSUPPORTED_OPERATION
);
}
}

private boolean isIngressCounter(UpfEntityType type) {
return type.equals(UpfEntityType.INGRESS_COUNTER);
}

private boolean isBiCounter(UpfEntityType type) {
return type.equals(UpfEntityType.COUNTER);
}

private boolean isEgressCounter(UpfEntityType type) {
return type.equals(UpfEntityType.EGRESS_COUNTER);
}
}
Loading

0 comments on commit fd3c3f0

Please sign in to comment.