diff --git a/src/main/java/org/opencord/kafka/integrations/DeviceKafkaIntegration.java b/src/main/java/org/opencord/kafka/integrations/DeviceKafkaIntegration.java
new file mode 100644
index 0000000..488a338
--- /dev/null
+++ b/src/main/java/org/opencord/kafka/integrations/DeviceKafkaIntegration.java
@@ -0,0 +1,140 @@
+/*
+ * Copyright 2018-present Open Networking Foundation
+ *
+ * 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.
+ */
+
+package org.opencord.kafka.integrations;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.device.DeviceEvent;
+import org.onosproject.net.device.DeviceListener;
+import org.onosproject.net.device.DeviceService;
+import org.onosproject.net.device.PortStatistics;
+import org.opencord.kafka.EventBusService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+import java.util.Iterator;
+import java.util.List;
+
+
+/**
+ * Listens for access device events and pushes them on a Kafka bus.
+ */
+@Component(immediate = true)
+public class DeviceKafkaIntegration {
+
+    public Logger log = LoggerFactory.getLogger(getClass());
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected EventBusService eventBusService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected DeviceService deviceService;
+
+    private final DeviceListener listener = new InternalDeviceListener();
+
+    private static final String TOPIC = "onos.kpis";
+
+    // event fields
+    private static final String TIMESTAMP = "timestamp";
+    private static final String DEVICE_ID = "deviceId";
+    private static final String PORTS = "ports";
+    private static final String PORT_ID = "portId";
+    private static final String PKT_RX = "pktRx";
+    private static final String PKT_TX = "pktTx";
+    private static final String BYTES_RX = "bytesRx";
+    private static final String BYTES_TX = "bytesTx";
+    private static final String PKT_RX_DROP = "pktRxDrp";
+    private static final String PKT_TX_DROP = "pktTxDrp";
+
+
+    @Activate
+    public void activate() {
+        deviceService.addListener(listener);
+        log.info("Started");
+    }
+
+    @Deactivate
+    public void deactivate() {
+        deviceService.removeListener(listener);
+        log.info("Stopped");
+    }
+
+    private void handle(List<PortStatistics> stats, DeviceId deviceId) {
+        eventBusService.send(TOPIC, serialize(stats, deviceId));
+    }
+
+    private JsonNode serialize(List<PortStatistics> stats, DeviceId deviceId) {
+
+        ObjectMapper mapper = new ObjectMapper();
+        ObjectNode kpis = mapper.createObjectNode();
+        ArrayNode ports = mapper.createArrayNode();
+
+        for (Iterator<PortStatistics> i = stats.iterator(); i.hasNext();) {
+            PortStatistics stat = i.next();
+
+            ObjectNode port = mapper.createObjectNode();
+            port.put(PORT_ID, stat.portNumber().toString());
+            port.put(PKT_RX, stat.packetsReceived());
+            port.put(PKT_TX, stat.packetsSent());
+            port.put(BYTES_RX, stat.bytesReceived());
+            port.put(BYTES_TX, stat.bytesSent());
+            port.put(PKT_RX_DROP, stat.packetsRxDropped());
+            port.put(PKT_TX_DROP, stat.packetsTxDropped());
+
+            ports.add(port);
+        }
+
+        kpis.put(TIMESTAMP, Instant.now().toString());
+        kpis.put(PORTS, ports);
+        kpis.put(DEVICE_ID, deviceId.toString());
+
+        return kpis;
+    }
+
+    private class InternalDeviceListener implements
+            DeviceListener {
+
+        @Override
+        public void event(DeviceEvent deviceEvent) {
+
+            if (deviceEvent.subject().manufacturer().contains("VOLTHA")) {
+                // TODO check the NNI port instead
+                return;
+            }
+
+            log.trace("Got DeviceEvent: " + deviceEvent.type());
+            switch (deviceEvent.type()) {
+            case PORT_STATS_UPDATED:
+                final DeviceId deviceId = deviceEvent.subject().id();
+                final List<PortStatistics> stats = deviceService.getPortStatistics(deviceId);
+                handle(stats, deviceId);
+                break;
+            default:
+                break;
+            }
+        }
+    }
+}