Add DAQClient and basic test to simulate access switch with device coupler

.github/workflows/tests.yml

@@ -202,3 +202,59 @@ jobs:
           cat inst/endpoint.log || true
           echo %%%%%%%%%%%%% Log lengths
           find inst -name \*.log | fgrep -v nodes/ | xargs wc -l
+
+  device_coupler_test:
+    needs: build_docker
+    runs-on: ubuntu-18.04
+    timeout-minutes: 40
+    strategy:
+      fail-fast: false
+    steps:
+    - uses: actions/download-artifact@v2
+      with:
+          name: build_artifacts_faucet
+          path: /tmp/build_artifacts_faucet
+    - uses: actions/download-artifact@v2
+      with:
+          name: build_artifacts_faux
+          path: /tmp/build_artifacts_faux
+    - uses: actions/download-artifact@v2
+      with:
+          name: build_artifacts_setup
+          path: /tmp/build_artifacts_setup
+    - uses: actions/download-artifact@v2
+      with:
+          name: build_artifacts_base
+          path: /tmp/build_artifacts_base
+    - uses: actions/checkout@v2
+      with:
+        fetch-depth: 0
+    - name: Set up Python 3.8
+      uses: actions/setup-python@v2
+      with:
+        python-version: 3.8
+    - name: Set up JDK 1.11
+      uses: actions/setup-java@v1
+      with:
+        java-version: 1.11
+    - name: Installing dependencies
+      env:
+        GIT_URL: ${{ secrets.GIT_URL }}
+      run: |
+        bin/setup_daq
+    - name: Loading docker images
+      run: |
+        bin/load_images faucet faux setup base
+    - name: Running device_coupler test
+      env:
+        DOCKER_STARTUP_TIMEOUT_MS: 60000
+        GCP_BASE64_CRED: ${{ secrets.GCP_BASE64_CRED }}
+        GCP_REFLECT_KEY_BASE64: ${{ secrets.GCP_REFLECT_KEY_BASE64 }}
+      run: |
+        device_coupler/testing/test_device_coupler
+    - name: Runtime logs
+      if: ${{ always() }}
+      run: |
+        echo %%%%%%% daq log cmdrun.log
+        cat inst/cmdrun.log || true
+        more inst/run-*/nodes/pass*/tmp/module_config.json | cat

bin/build_device_coupler

@@ -0,0 +1,9 @@
+#!/bin/bash -e
+
+image_name="device_coupler"
+
+echo Starting $image_name build...
+
+docker build . -f docker/modules/Dockerfile.device_coupler -t daqf/$image_name
+
+echo Done with $image_name build.

bin/build_shard

@@ -1,6 +1,6 @@
 #!/bin/bash -e
 
-setup=aardvark,aardvark2,default,networking,switch
+setup=aardvark,aardvark2,default,networking,switch,device_coupler
 faux=faux1,faux2
 faucet=faucet,gauge
 base=test_bacnet,test_mudgee,test_brute,test_discover,usi,test_ipaddr

bin/net_clean

@@ -1,7 +1,7 @@
 #!/bin/bash
 
 echo Cleaning docker images...
-for image in daqf/faucet daqf/gauge daqf/aardvark; do
+for image in daqf/faucet daqf/gauge daqf/aardvark daqf/device_coupler; do
     echo Cleaning docker image $image...
     images=$(docker ps --filter ancestor=$image -q)
     if [ -n "$images" ]; then

bin/setup_device_coupler

@@ -0,0 +1,34 @@
+#!/bin/bash -e
+
+ROOT=$(realpath $(dirname $0)/..)
+echo ROOT is $ROOT
+
+usage () {
+  echo " $0 -t <trunk-interface>" 1>&2;
+  exit 1
+}
+
+while getopts "t:" argch; do
+  case "${argch}" in
+    t)
+      trunk_port=${OPTARG}
+      ;;
+    *)
+      usage
+      ;;
+  esac
+done
+shift "$((OPTIND-1))"
+
+if [ -z "${trunk_port}" ]; then
+  usage
+fi
+
+$ROOT/bin/build_device_coupler
+$ROOT/bin/start_device_coupler
+
+pid=$(docker inspect --format="{{ .State.Pid  }}" dev_coupler_box)
+sudo ip link set $trunk_port netns $pid
+
+docker exec dev_coupler_box ip link set $trunk_port up
+docker exec dev_coupler_box python3 device_coupler/network_helper.py --bridge dev_br0 --trunk-iface $trunk_port

bin/start_device_coupler

@@ -0,0 +1,20 @@
+#!/bin/bash -e
+
+ROOT=$(realpath $(dirname $0)/..)
+CONTAINER=dev_coupler_box
+INSTDIR=$ROOT/inst
+IMAGE=daqf/device_coupler
+# TODO: Write log files here
+mkdir -p $INSTDIR/device_coupler
+docker_volumes+="\
+        -v $INSTDIR:/root/inst"
+
+if docker container inspect $CONTAINER --format '{{ .Name }}' > /dev/null 2>&1; then
+    echo -n "Cleansing old container $CONTAINER"
+    docker rm -f $CONTAINER
+fi
+
+cid=$(docker run -d --privileged --name $CONTAINER \
+             $docker_volumes $IMAGE)
+
+echo Container $CONTAINER started as $cid

cmd/faux

@@ -86,7 +86,7 @@ fi
 echo Launching faux $FAUX_ARGS...
 
 if docker container inspect $CONTAINER --format '{{ .Name }}' > /dev/null 2>&1; then
-    echo -n "Clensing old container "
+    echo -n "Cleansing old container "
     docker rm -f $CONTAINER
 fi
 

daq/runner.py

@@ -306,7 +306,9 @@ def _on_session_start(self, request):
             assert remote_ip, 'remote request ip not specified'
             device = self._devices.create_if_absent(request.device_mac)
             device.port.flapping_start = None  # In case this was set from last disconnect.
-            device.dhcp_mode = DhcpMode.EXTERNAL
+            # Assigned VLAN set => FOT mode.
+            # For device coupler mode, DHCP mode shouldn't be external.
+            device.dhcp_mode = DhcpMode.EXTERNAL if request.assigned_vlan else DhcpMode.NORMAL
             device.session_endpoint = request.endpoint
             self._remote_trigger(device, request.device_vlan, request.assigned_vlan)
             self._udmi.discovery(device)
@@ -517,7 +519,7 @@ def _handle_device_learn(self, target_mac, vid):
             self._target_set_trigger(device)
 
     def _remote_trigger(self, device, device_vlan, assigned_vlan):
-        assert device_vlan and assigned_vlan, 'expected both device_ and assigned_ vlans'
+        assert device_vlan, 'expected device_vlan'
         device.port.flapping_start = 0
         device.port.active = True
 

daq/topology.py

@@ -46,6 +46,8 @@ class FaucetTopology:
     _OFPP_IN_PORT = 0xfffffff8
     _DOT1X_ETH_TYPE = 0x888e
     _DEFAULT_GAUGE_VARZ_PORT = 9303
+    _VXLAN_ACL = 'vxlan'
+    _COUPLER_ACL = 'vxlan_coupler'
 
     def __init__(self, config):
         self.config = config
@@ -172,14 +174,20 @@ def direct_device_traffic(self, device):
         LOGGER.info('Direct device %s traffic to %s %s', device.mac, device.port.vxlan, port_set)
         if device.port.vxlan:
             egress_vlan = device.assigned if device.assigned else self._egress_vlan
-            sec_topology['interfaces'][self.VXLAN_SEC_TRUNK_PORT] = (
-                self._make_sec_trunk_interface(addition=(egress_vlan,)))
+            if egress_vlan:
+                sec_topology['interfaces'][self.VXLAN_SEC_TRUNK_PORT] = (
+                    self._make_sec_trunk_interface(addition=(egress_vlan,)))
 
             if port_set:
                 interface = sec_interfaces.setdefault(device.port.vxlan, {})
-                interface['tagged_vlans'] = [vlan, egress_vlan]
+                if egress_vlan:
+                    interface['tagged_vlans'] = [vlan, egress_vlan]
+                    incoming_acl = self._VXLAN_ACL
+                else:
+                    interface['native_vlan'] = vlan
+                    incoming_acl = self._COUPLER_ACL
                 interface['name'] = str(device)
-                interface['acl_in'] = self.INCOMING_ACL_FORMAT % 'vxlan'
+                interface['acl_in'] = self.INCOMING_ACL_FORMAT % incoming_acl
             else:
                 sec_interfaces.pop(device.port.vxlan, None)
 
@@ -523,7 +531,10 @@ def _generate_main_acls(self):
         self._add_acl_rule(secondary_acl, allow=1)
         acls[self.INCOMING_ACL_FORMAT % self.sec_name] = secondary_acl
 
-        acls[self.INCOMING_ACL_FORMAT % 'vxlan'] = [self._make_acl_rule(ports=[1])]
+        acls[self.INCOMING_ACL_FORMAT % self._VXLAN_ACL] = [
+            self._make_acl_rule(ports=[self.VXLAN_SEC_TRUNK_PORT])]
+        acls[self.INCOMING_ACL_FORMAT % self._COUPLER_ACL] = [
+            self._make_acl_rule(ports=[self.VXLAN_SEC_TRUNK_PORT], allow=True)]
 
         for port_set in range(1, self.sec_port):
             vlan = self._port_set_vlan(port_set)

device_coupler/daq_client.py

@@ -0,0 +1,123 @@
+"""Module to send gRPC requests to DAQ and manage test sessions"""
+
+from __future__ import absolute_import
+
+import grpc
+import threading
+import traceback
+
+from device_coupler.utils import get_logger
+from device_coupler.ovs_helper import OvsHelper
+
+from daq.proto.session_server_pb2 import SessionParams, SessionResult
+from daq.proto.session_server_pb2_grpc import SessionServerStub
+
+
+DEFAULT_SERVER_ADDRESS = '127.0.0.1'
+CONNECT_TIMEOUT_SEC = 60
+
+
+class DAQClient():
+    """gRPC client to send device result"""
+
+    def __init__(self, target, tunnel_ip, ovs_bridge):
+        self._logger = get_logger('daqclient')
+        self._logger.info('Using target %s', target)
+        self._channel = grpc.insecure_channel(target)
+        self._stub = None
+        self._mac_sessions = {}
+        self._lock = threading.Lock()
+        self._tunnel_ip = tunnel_ip
+        self._endpoint_handler = OvsHelper()
+        self._ovs_bridge = ovs_bridge
+        # Assigned VLAN is always set to 0 for non-FOT DAQ Client
+        self._assigned_vlan = 0
+
+    def start(self):
+        """Start the client handler"""
+        grpc.channel_ready_future(self._channel).result(timeout=CONNECT_TIMEOUT_SEC)
+        self._stub = SessionServerStub(self._channel)
+
+    def stop(self):
+        """Stop client handler"""
+
+    def _connect(self, mac, vlan):
+        self._logger.info('Connecting %s with VLAN %s', mac, vlan)
+        session_params = SessionParams()
+        session_params.device_mac = mac
+        session_params.device_vlan = vlan
+        session_params.assigned_vlan = self._assigned_vlan
+        session_params.endpoint.ip = self._tunnel_ip or DEFAULT_SERVER_ADDRESS
+        session = self._stub.StartSession(session_params)
+        thread = threading.Thread(target=lambda: self._run_test_session(mac, session))
+        thread.start()
+        self._logger.info('Connection of %s with VLAN %s succeeded', mac, vlan)
+        return session
+
+    def disconnect(self, mac):
+        with self._lock:
+            session = self._mac_sessions.get(mac, {}).get('session')
+            if session:
+                session.cancel()
+                mac_session = self._mac_sessions.pop(mac)
+                index = mac_session['index']
+                interface = "vxlan%s" % index
+                self._endpoint_handler.remove_vxlan_endpoint(interface, self._ovs_bridge)
+                self._logger.info('Session terminated for %s', mac)
+            else:
+                self._logger.warning('Attempt to disconnect unconnected device %s', mac)
+
+    def _is_session_running(self, mac, progress):
+        result = self._process_session_progress(mac, progress)
+        return result not in ('PASSED', 'FAILED', 'ERROR')
+
+    def _process_session_progress(self, mac, progress):
+        endpoint = progress.endpoint
+        result_code = progress.result.code
+        assert not (endpoint.ip and result_code), 'both endpoint.ip and result.code defined'
+        if result_code:
+            result_name = SessionResult.ResultCode.Name(result_code)
+            self._logger.info('Device report %s as %s', mac, result_name)
+            return result_name
+        if endpoint.ip:
+            self._logger.info('Device report %s endpoint ip: %s)',
+                              mac, endpoint.ip)
+            # TODO: Change the way indexes work. Check for VXLAN port being sent
+            index = endpoint.vni
+            device = self._mac_sessions[mac]
+            device['index'] = index
+            interface = "vxlan%s" % index
+            self._endpoint_handler.remove_vxlan_endpoint(interface, self._ovs_bridge)
+            self._endpoint_handler.create_vxlan_endpoint(interface, endpoint.ip, index)
+            self._endpoint_handler.add_iface_to_bridge(
+                self._ovs_bridge, interface, tag=device['device_vlan'])
+        return None
+
+    def _run_test_session(self, mac, session):
+        try:
+            for progress in session:
+                if not self._is_session_running(mac, progress):
+                    break
+            self._logger.info('Progress complete for %s', mac)
+        except Exception as e:
+            self._logger.error('Progress exception: %s', e)
+            self._logger.error('Traceback: %s', traceback.format_exc())
+        self.disconnect(mac)
+
+    def _initiate_test_session(self, mac, device_vlan):
+        if mac in self._mac_sessions:
+            self._logger.info('Test session for %s already exists. Ignoring.', mac)
+            return
+        self._logger.info('Initiating test session for %s on VLAN %s', mac, device_vlan)
+
+        if device_vlan:
+            self._mac_sessions[mac] = {}
+            self._mac_sessions[mac]['device_vlan'] = device_vlan
+            self._mac_sessions[mac]['session'] = self._connect(mac, device_vlan)
+        self._logger.info('Initiated test session %s', self._mac_sessions[mac])
+
+    def process_device_discovery(self, mac, device_vlan):
+        """Process discovery of device to be tested"""
+        # TODO: End existing test session and start new one if discovered on another port
+        with self._lock:
+            self._initiate_test_session(mac, device_vlan)

device_coupler/daq_config/daq_config.yaml

@@ -0,0 +1,16 @@
+# DAQ FOT configuration.
+
+include: ${DAQ_LIB}/config/system/default.yaml
+
+host_tests: ${DAQ_LIB}/config/modules/dts.conf
+
+switch_setup:
+  lo_port: 6653
+  varz_port: 5678
+
+run_trigger:
+  vlan_start: 209
+  vlan_end: 211
+
+device_reporting:
+  server_port: 50051

device_coupler/daq_config/startup.yaml

@@ -0,0 +1,4 @@
+site_description: generated
+switch_setup:
+  endpoint:
+    ip: 10.0.0.1

device_coupler/device_coupler_go

@@ -0,0 +1,7 @@
+#!/bin/bash -e
+
+# Start OVS
+sudo /usr/share/openvswitch/scripts/ovs-ctl start
+
+# Block
+tail -f

device_coupler/make_daq_request

@@ -0,0 +1,10 @@
+#!/bin/bash -e
+
+tunnel_ip=$(ip addr show eth0 | fgrep inet | sed -nr 's~.*inet ([0-9.]+)/.*~\1~p')
+grpc_port=50051
+target=$(echo $tunnel_ip | sed 's:[^.]*$:1:'):$grpc_port
+device_mac="9a:02:57:1e:8f:01"
+device_vlan=210
+ovs_br="dev_br0"
+
+python3 device_coupler/test_run.py --target $target --source $tunnel_ip --device_mac $device_mac --device_vlan $device_vlan --ovs_br $ovs_br