packet-out escapes twisted thread
Change-Id: I5f1df9d741a7e1861d7a70c5d66584d2eff2b1bf
diff --git a/voltha/adapters/openolt/openolt_device.py b/voltha/adapters/openolt/openolt_device.py
index d3c55f9..2bc036a 100644
--- a/voltha/adapters/openolt/openolt_device.py
+++ b/voltha/adapters/openolt/openolt_device.py
@@ -25,6 +25,7 @@
from voltha.adapters.openolt.openolt_utils import OpenoltUtils
from voltha.adapters.openolt.openolt_grpc import OpenoltGrpc
from voltha.adapters.openolt.openolt_indications import OpenoltIndications
+from voltha.adapters.openolt.openolt_packet import OpenoltPacket
from voltha.adapters.openolt.openolt_kafka_admin import KAdmin
@@ -111,6 +112,8 @@
def do_state_init(self, event):
self.log.debug('init')
+ self._packet = OpenoltPacket(self)
+ self._packet.start()
self._indications = OpenoltIndications(self)
self._indications.start()
diff --git a/voltha/adapters/openolt/openolt_grpc.py b/voltha/adapters/openolt/openolt_grpc.py
index 54c0500..a7cfd07 100644
--- a/voltha/adapters/openolt/openolt_grpc.py
+++ b/voltha/adapters/openolt/openolt_grpc.py
@@ -20,8 +20,7 @@
import time
from simplejson import dumps
from twisted.internet import reactor
-from google.protobuf.json_format import MessageToJson
-from voltha.northbound.kafka.kafka_proxy import get_kafka_proxy
+from voltha.northbound.kafka.kafka_proxy import kafka_send_pb
from voltha.adapters.openolt.protos import openolt_pb2_grpc, openolt_pb2
@@ -53,22 +52,6 @@
def indications_thread(self):
- def forward_indication(topic, msg):
- try:
- self.log.debug('forward indication', topic=topic, msg=msg)
- kafka_proxy = get_kafka_proxy()
- if kafka_proxy and not kafka_proxy.is_faulty():
- self.log.debug('kafka-proxy-available')
- kafka_proxy.send_message(
- topic,
- dumps(MessageToJson(
- msg,
- including_default_value_fields=True)))
- else:
- self.log.error('kafka-proxy-unavailable')
- except Exception, e:
- self.log.exception('failed-sending-message', e=e)
-
self.log.debug('openolt grpc connecting to olt')
self.stub = openolt_pb2_grpc.OpenoltStub(self.channel)
@@ -126,4 +109,4 @@
indications=ind)
continue
- forward_indication("openolt.ind", ind)
+ kafka_send_pb("openolt.ind", ind)
diff --git a/voltha/adapters/openolt/openolt_packet.py b/voltha/adapters/openolt/openolt_packet.py
new file mode 100644
index 0000000..cfd838c
--- /dev/null
+++ b/voltha/adapters/openolt/openolt_packet.py
@@ -0,0 +1,134 @@
+#
+# Copyright 2019 the original author or authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import threading
+from google.protobuf.json_format import Parse
+from simplejson import loads
+import structlog
+from scapy.layers.l2 import Ether, Dot1Q
+import binascii
+
+from common.frameio.frameio import hexify
+from voltha.protos.openflow_13_pb2 import PacketOut
+from voltha.adapters.openolt.openolt_kafka_consumer import KConsumer
+from voltha.core.flow_decomposer import OUTPUT
+from voltha.protos.device_pb2 import Port
+from voltha.adapters.openolt.protos import openolt_pb2
+
+
+class OpenoltPacket(object):
+ def __init__(self, device):
+ self.log = structlog.get_logger()
+ self.device = device
+ self.packet_thread_handle = threading.Thread(
+ target=self.packet_thread)
+ self.packet_thread_handle.setDaemon(True)
+
+ def start(self):
+ self.packet_thread_handle.start()
+
+ def stop(self):
+ pass
+
+ def packet_thread(self):
+ self.log.debug('openolt packet-out thread starting')
+ KConsumer(self.packet_process, 'voltha.pktout')
+
+ def packet_process(self, topic, msg):
+
+ def get_port_out(opo):
+ for action in opo.actions:
+ if action.type == OUTPUT:
+ return action.output.port
+
+ pb = Parse(loads(msg), PacketOut(), ignore_unknown_fields=True)
+
+ logical_device_id = pb.id
+ ofp_packet_out = pb.packet_out
+
+ self.log.debug("received packet-out form kafka",
+ logical_device_id=logical_device_id,
+ ofp_packet_out=ofp_packet_out)
+
+ egress_port = get_port_out(ofp_packet_out)
+ msg = ofp_packet_out.data
+
+ self.log.debug('rcv-packet-out', logical_device_id=logical_device_id,
+ egress_port=egress_port,
+ # adapter_name=self.adapter_name,
+ data=hexify(msg))
+
+ pkt = Ether(msg)
+ self.log.debug('packet out', egress_port=egress_port,
+ packet=str(pkt).encode("HEX"))
+
+ # Find port type
+ egress_port_type = self.device.platform \
+ .intf_id_to_port_type_name(egress_port)
+
+ if egress_port_type == Port.ETHERNET_UNI:
+
+ if pkt.haslayer(Dot1Q):
+ outer_shim = pkt.getlayer(Dot1Q)
+ if isinstance(outer_shim.payload, Dot1Q):
+ # If double tag, remove the outer tag
+ payload = (
+ Ether(src=pkt.src, dst=pkt.dst,
+ type=outer_shim.type) /
+ outer_shim.payload
+ )
+ else:
+ payload = pkt
+ else:
+ payload = pkt
+
+ send_pkt = binascii.unhexlify(str(payload).encode("HEX"))
+
+ self.log.debug(
+ 'sending-packet-to-ONU', egress_port=egress_port,
+ intf_id=self.device.platform.intf_id_from_uni_port_num(
+ egress_port),
+ onu_id=self.device.platform.onu_id_from_port_num(egress_port),
+ uni_id=self.device.platform.uni_id_from_port_num(egress_port),
+ port_no=egress_port,
+ packet=str(payload).encode("HEX"))
+
+ onu_pkt = openolt_pb2.OnuPacket(
+ intf_id=self.device.platform.intf_id_from_uni_port_num(
+ egress_port),
+ onu_id=self.device.platform.onu_id_from_port_num(egress_port),
+ port_no=egress_port,
+ pkt=send_pkt)
+
+ self.device._grpc.stub.OnuPacketOut(onu_pkt)
+
+ elif egress_port_type == Port.ETHERNET_NNI:
+ self.log.debug('sending-packet-to-uplink', egress_port=egress_port,
+ packet=str(pkt).encode("HEX"))
+
+ send_pkt = binascii.unhexlify(str(pkt).encode("HEX"))
+
+ uplink_pkt = openolt_pb2.UplinkPacket(
+ intf_id=self.device.platform.intf_id_from_nni_port_num(
+ egress_port),
+ pkt=send_pkt)
+
+ self.device._grpc.stub.UplinkPacketOut(uplink_pkt)
+
+ else:
+ self.log.warn('Packet-out-to-this-interface-type-not-implemented',
+ egress_port=egress_port,
+ port_type=egress_port_type)
diff --git a/voltha/core/local_handler.py b/voltha/core/local_handler.py
index 39f550e..6daef6e 100644
--- a/voltha/core/local_handler.py
+++ b/voltha/core/local_handler.py
@@ -41,6 +41,7 @@
from voltha.protos.omci_alarm_db_pb2 import AlarmDeviceData
from requests.api import request
from common.utils.asleep import asleep
+from voltha.northbound.kafka.kafka_proxy import kafka_send_pb
log = structlog.get_logger()
@@ -1116,8 +1117,17 @@
agent = self.core.get_logical_device_agent(packet_out.id)
agent.packet_out(packet_out.packet_out)
- for request in request_iterator:
- forward_packet_out(packet_out=request)
+ for req in request_iterator:
+ device_agent = self.core.get_logical_device_agent(req.id)
+ adapter_name = device_agent.device_adapter_agent.name
+
+ if adapter_name == 'openolt':
+ log.debug('fast path pkt-out to kafka')
+ # topic = 'openolt.pktout:{}'.format(req.id)
+ topic = 'voltha.pktout'
+ kafka_send_pb(topic, req)
+ else:
+ forward_packet_out(packet_out=req)
log.debug('stop-stream-packets-out')
diff --git a/voltha/northbound/kafka/kafka_proxy.py b/voltha/northbound/kafka/kafka_proxy.py
index 0fba534..08d42f1 100644
--- a/voltha/northbound/kafka/kafka_proxy.py
+++ b/voltha/northbound/kafka/kafka_proxy.py
@@ -15,18 +15,20 @@
#
-from confluent_kafka import Producer as _kafkaProducer
+import threading
from structlog import get_logger
from twisted.internet import reactor
from twisted.internet.defer import inlineCallbacks, returnValue
from twisted.internet.threads import deferToThread
from zope.interface import implementer
+from google.protobuf.json_format import MessageToJson
+from simplejson import dumps
+from confluent_kafka import Producer as _kafkaProducer
+from confluent_kafka import Consumer, KafkaError
from common.utils.consulhelpers import get_endpoint_from_consul
from voltha.northbound.kafka.event_bus_publisher import EventBusPublisher
from voltha.registry import IComponent
-from confluent_kafka import Consumer, KafkaError
-import threading
log = get_logger()
@@ -336,3 +338,19 @@
# Common method to get the singleton instance of the kafka proxy class
def get_kafka_proxy():
return KafkaProxy._kafka_instance
+
+def kafka_send_pb(topic, msg):
+ try:
+ log.debug('send protobuf to kafka', topic=topic, msg=msg)
+ kafka_proxy = get_kafka_proxy()
+ if kafka_proxy and not kafka_proxy.is_faulty():
+ log.debug('kafka-proxy-available')
+ kafka_proxy.send_message(
+ topic,
+ dumps(MessageToJson(
+ msg,
+ including_default_value_fields=True)))
+ else:
+ log.error('kafka-proxy-unavailable')
+ except Exception, e:
+ log.exception('failed-sending-protobuf', e=e)