源码解读ODL的MAC地址学习(二)

address-tracker.yang：
module address-tracker {
  ...... 
  grouping address-node-connector {
    list addresses {
      key id;
      leaf id {
        description "A 64-bit key for this observation. This is opaque and should not be interpreted.";
        type uint64;
      }
      leaf mac {
        type yang:mac-address;
        description "MAC address";
      }
      leaf ip {
        type inet:ip-address;
        description "IPv4 or IPv6 address";
      }
      leaf vlan {
        type ethernet:vlan-id;
        description "VLAN id";
      }
      leaf first-seen {
        type int64;
        description "Timestamp (number of ms since January 1, 1970, 00:00:00 GMT) of observing this address for the first time";
      }
      leaf last-seen {
        type int64;
        description "The most recent timestamp (tnumber of ms since January 1, 1970, 00:00:00 GMT) of observing this address";
      }
    }
  }
  augment "/inv:nodes/inv:node/inv:node-connector" {
    ext:augment-identifier "address-capable-node-connector";
    uses address-node-connector;
  }
}

public class AddressObserverUsingArp implements ArpPacketListener {
  public AddressObserverUsingArp(org.opendaylight.l2switch.addresstracker.addressobserver.AddressObservationWriter addressObservationWriter) {
    this.addressObservationWriter = addressObservationWriter;
  }
  @Override
  public void onArpPacketReceived(ArpPacketReceived packetReceived) {
    if(packetReceived == null || packetReceived.getPacketChain() == null) {
      return;
    }
    RawPacket rawPacket = null;   //原始数据包
    EthernetPacket ethernetPacket = null;  //以太网报文
    ArpPacket arpPacket = null;   //arp报文
 //得到相应的原始数据包、以太网报文和arp报文
    for(PacketChain packetChain : packetReceived.getPacketChain()) {
      if(packetChain.getPacket() instanceof RawPacket) {
        rawPacket = (RawPacket) packetChain.getPacket();
      } else if(packetChain.getPacket() instanceof EthernetPacket) {
        ethernetPacket = (EthernetPacket) packetChain.getPacket();
      } else if(packetChain.getPacket() instanceof ArpPacket) {
        arpPacket = (ArpPacket) packetChain.getPacket();
      }
    }
    if(rawPacket == null || ethernetPacket == null || arpPacket == null) {
      return;
    }
    //调用addressObservationWriter类的addAddress()方法
    addressObservationWriter.addAddress(ethernetPacket.getSourceMac(),
        new IpAddress(arpPacket.getSourceProtocolAddress().toCharArray()),
        rawPacket.getIngress());
  }
}

public class AddressObservationWriter {
  ......
  public void setTimestampUpdateInterval(long timestampUpdateInterval) {
    this.timestampUpdateInterval = timestampUpdateInterval;  //定义时间戳
  }
  //向datastore中添加Address
  public void addAddress(MacAddress macAddress, IpAddress ipAddress, NodeConnectorRef nodeConnectorRef) {
    if(macAddress == null || ipAddress == null || nodeConnectorRef == null) {
      return;
    }
    // 锁住线程，当前写入线程未结束时不能进行下一个写入线程
    NodeConnectorLock nodeConnectorLock;
    synchronized(this) {
      nodeConnectorLock = lockMap.get(nodeConnectorRef);
      if(nodeConnectorLock == null) {
        nodeConnectorLock = new NodeConnectorLock();
        lockMap.put(nodeConnectorRef, nodeConnectorLock);
      }
    }
    synchronized(nodeConnectorLock) {
      // Ensure previous transaction finished writing to the db
      CheckedFuture<Void, TransactionCommitFailedException> future = futureMap.get(nodeConnectorLock);
      if (future != null) {
        try {
          future.get();
        }
        catch (InterruptedException|ExecutionException e) {
          _logger.error("Exception while waiting for previous transaction to finish", e);
        }
      }
      //初始化addressBuilder
      long now = new Date().getTime();
      final AddressCapableNodeConnectorBuilder acncBuilder = new AddressCapableNodeConnectorBuilder();
      final AddressesBuilder addressBuilder = new AddressesBuilder()
          .setIp(ipAddress)
          .setMac(macAddress)
          .setFirstSeen(now)
          .setLastSeen(now);
      List<Addresses> addresses = null;
      // 从datastore中读取现有的Address
      ReadOnlyTransaction readTransaction = dataService.newReadOnlyTransaction();
      NodeConnector nc = null;
      try {
        Optional<NodeConnector> dataObjectOptional = readTransaction.read(LogicalDatastoreType.OPERATIONAL, (InstanceIdentifier<NodeConnector>) nodeConnectorRef.getValue()).get();
        if(dataObjectOptional.isPresent())
          nc = (NodeConnector) dataObjectOptional.get();
      } catch(Exception e) {
        _logger.error("Error reading node connector {}", nodeConnectorRef.getValue());
        readTransaction.close();
        throw new RuntimeException("Error reading from operational store, node connector : " + nodeConnectorRef, e);
      }
      readTransaction.close();
      if(nc == null) {
        return;
      }
      AddressCapableNodeConnector acnc = (AddressCapableNodeConnector) nc.getAugmentation(AddressCapableNodeConnector.class);
      // Address存在的情况
      if(acnc != null && acnc.getAddresses() != null) {
        //从现有的Address中查找相应的MAC-IP对，并且更新last-seen时间戳
        addresses = acnc.getAddresses();
        for(int i = 0; i < addresses.size(); i++) {
          if(addresses.get(i).getIp().equals(ipAddress) && addresses.get(i).getMac().equals(macAddress)) {
            if((now - addresses.get(i).getLastSeen()) > timestampUpdateInterval) {
              addressBuilder.setFirstSeen(addresses.get(i).getFirstSeen())
                  .setKey(addresses.get(i).getKey());
              addresses.remove(i);
              break;
            } else {
              return;
            }
          }
        }
      }
      //Address不存在的情况，需要写入新的Address
      else {
        addresses = new ArrayList<>();
      }
      if(addressBuilder.getKey() == null) {
        addressBuilder.setKey(new AddressesKey(BigInteger.valueOf(addressKey.getAndIncrement())));
      }
      addresses.add(addressBuilder.build());
      acncBuilder.setAddresses(addresses);
      //定义写入datastore的IID是AddressCapableNodeConnector
      InstanceIdentifier<AddressCapableNodeConnector> addressCapableNcInstanceId =
          ((InstanceIdentifier<NodeConnector>) nodeConnectorRef.getValue())
              .augmentation(AddressCapableNodeConnector.class);
      final WriteTransaction writeTransaction = dataService.newWriteOnlyTransaction();
      // 向datastore中写入Address，其中datastore的类型是operational，IID是AddressCapableNodeConnector
      writeTransaction.merge(LogicalDatastoreType.OPERATIONAL, addressCapableNcInstanceId, acncBuilder.build());
      final CheckedFuture writeTxResultFuture = writeTransaction.submit();
      Futures.addCallback(writeTxResultFuture, new FutureCallback() {
        @Override
        public void onSuccess(Object o) {
          _logger.debug("AddressObservationWriter write successful for tx :{}", writeTransaction.getIdentifier());
        }
        @Override
        public void onFailure(Throwable throwable) {
          _logger.error("AddressObservationWriter write transaction {} failed", writeTransaction.getIdentifier(), throwable.getCause());
        }
      });
      futureMap.put(nodeConnectorLock, writeTxResultFuture);
    }
  }
}

public class FlowWriterServiceImpl implements FlowWriterService {
  ......
  //添加MAC-MAC流表
  @Override
  public void addMacToMacFlow(MacAddress sourceMac, MacAddress destMac, NodeConnectorRef destNodeConnectorRef) {
    Preconditions.checkNotNull(destMac, "Destination mac address should not be null.");
    Preconditions.checkNotNull(destNodeConnectorRef, "Destination port should not be null.");
    // 如果两个MAC地址相同则不添加流表
    if(sourceMac != null && destMac.equals(sourceMac)) {
      _logger.info("In addMacToMacFlow: No flows added. Source and Destination mac are same.");
      return;
    }
    // 生成table ID
    TableKey flowTableKey = new TableKey((short) flowTableId);
    //build a flow path based on node connector to program flow
    InstanceIdentifier<Flow> flowPath = buildFlowPath(destNodeConnectorRef, flowTableKey);
    //生成流表项的主体
    Flow flowBody = createMacToMacFlow(flowTableKey.getId(), flowPriority, sourceMac, destMac, destNodeConnectorRef);
    //向配置数据中提交流表项
    writeFlowToConfigData(flowPath, flowBody);
  }
  //生成对称的两条流表项
  public void addBidirectionalMacToMacFlows(MacAddress sourceMac,
                                            NodeConnectorRef sourceNodeConnectorRef,
                                            MacAddress destMac,
                                            NodeConnectorRef destNodeConnectorRef) {
    Preconditions.checkNotNull(sourceMac, "Source mac address should not be null.");
    Preconditions.checkNotNull(sourceNodeConnectorRef, "Source port should not be null.");
    Preconditions.checkNotNull(destMac, "Destination mac address should not be null.");
    Preconditions.checkNotNull(destNodeConnectorRef, "Destination port should not be null.");
    if(sourceNodeConnectorRef.equals(destNodeConnectorRef)) {
      _logger.info("In addMacToMacFlowsUsingShortestPath: No flows added. Source and Destination ports are same.");
      return;
    }
    // 在源端口添加目的MAC-源MAC的流表项
    addMacToMacFlow(destMac, sourceMac, sourceNodeConnectorRef);
    // 在目的端口添加目的MAC-源MAC的流表项
    addMacToMacFlow(sourceMac, destMac, destNodeConnectorRef);
  }
  private InstanceIdentifier<Flow> buildFlowPath(NodeConnectorRef nodeConnectorRef, TableKey flowTableKey) {
    // 生成流表项的ID
    FlowId flowId = new FlowId(String.valueOf(flowIdInc.getAndIncrement()));
    FlowKey flowKey = new FlowKey(flowId);
   //生成流表的IID
    return InstanceIdentifierUtils.generateFlowInstanceIdentifier(nodeConnectorRef, flowTableKey, flowKey);
  }
  //具体生成一条mac-mac流表项
  private Flow createMacToMacFlow(Short tableId, int priority,
                                  MacAddress sourceMac, MacAddress destMac, NodeConnectorRef destPort) {
    // 定义流表项实例
    FlowBuilder macToMacFlow = new FlowBuilder() //
        .setTableId(tableId) //
        .setFlowName("mac2mac");
    // 用自己的哈希码作为流表项的id
    macToMacFlow.setId(new FlowId(Long.toString(macToMacFlow.hashCode())));
    // 生成匹配字段的实例
 //匹配字段为目的mac
    EthernetMatchBuilder ethernetMatchBuilder = new EthernetMatchBuilder() //
        .setEthernetDestination(new EthernetDestinationBuilder() //
            .setAddress(destMac) //
            .build());
    //如果源mac存在的话添加源mac匹配
    if(sourceMac != null) {
      ethernetMatchBuilder.setEthernetSource(new EthernetSourceBuilder()
          .setAddress(sourceMac)
          .build());
    }
    EthernetMatch ethernetMatch = ethernetMatchBuilder.build();
    Match match = new MatchBuilder()
        .setEthernetMatch(ethernetMatch)
        .build();
    //得到目的port的ID
    Uri destPortUri = destPort.getValue().firstKeyOf(NodeConnector.class, NodeConnectorKey.class).getId();
 //定义output action
    Action outputToControllerAction = new ActionBuilder() //
        .setOrder(0)
        .setAction(new OutputActionCaseBuilder() //
            .setOutputAction(new OutputActionBuilder() //
                .setMaxLength(0xffff) //
                .setOutputNodeConnector(destPortUri) //定义目的端口为output
                .build()) //
            .build()) //
        .build();
    // 生成action的应用
    ApplyActions applyActions = new ApplyActionsBuilder().setAction(ImmutableList.of(outputToControllerAction))
        .build();
    //将这个action的应用添加到指令中
    Instruction applyActionsInstruction = new InstructionBuilder() //
        .setOrder(0)
        .setInstruction(new ApplyActionsCaseBuilder()//
            .setApplyActions(applyActions) //
            .build()) //
        .build();
    // 生成流表实例
    macToMacFlow
        .setMatch(match) //
        .setInstructions(new InstructionsBuilder() //
            .setInstruction(ImmutableList.of(applyActionsInstruction)) //
            .build()) //
        .setPriority(priority) //
        .setBufferId(OFConstants.OFP_NO_BUFFER) //
        .setHardTimeout(flowHardTimeout) //
        .setIdleTimeout(flowIdleTimeout) //
        .setCookie(new FlowCookie(BigInteger.valueOf(flowCookieInc.getAndIncrement())))
        .setFlags(new FlowModFlags(false, false, false, false, false));
    return macToMacFlow.build();
  }
  //向配置数据中提交流表项
  private Future<RpcResult<AddFlowOutput>> writeFlowToConfigData(InstanceIdentifier<Flow> flowPath,
                                                                 Flow flow) {
    final InstanceIdentifier<Table> tableInstanceId = flowPath.<Table>firstIdentifierOf(Table.class);
    final InstanceIdentifier<Node> nodeInstanceId = flowPath.<Node>firstIdentifierOf(Node.class);
    final AddFlowInputBuilder builder = new AddFlowInputBuilder(flow);
    builder.setNode(new NodeRef(nodeInstanceId));
    builder.setFlowRef(new FlowRef(flowPath));
    builder.setFlowTable(new FlowTableRef(tableInstanceId));
    builder.setTransactionUri(new Uri(flow.getId().getValue()));
    return salFlowService.addFlow(builder.build());
  }
}