diff --git a/internal/pkg/afrouterd/coreMonitor.go b/internal/pkg/afrouterd/coreMonitor.go
new file mode 100644
index 0000000..6a86c64
--- /dev/null
+++ b/internal/pkg/afrouterd/coreMonitor.go
@@ -0,0 +1,170 @@
+/*
+ * 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 afrouterd
+
+import (
+	"fmt"
+	"github.com/golang/protobuf/ptypes/empty"
+	"github.com/opencord/voltha-go/common/log"
+	pb "github.com/opencord/voltha-protos/go/afrouter"
+	cmn "github.com/opencord/voltha-protos/go/common"
+	vpb "github.com/opencord/voltha-protos/go/voltha"
+	"golang.org/x/net/context"
+	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"time"
+)
+
+func getVolthaPods(cs *kubernetes.Clientset) ([]*volthaPod, error) {
+	pods, err := cs.CoreV1().Pods(podNamespace).List(metav1.ListOptions{LabelSelector: podLabelSelector})
+	if err != nil {
+		return nil, err
+	}
+
+	var rwPods []*volthaPod
+items:
+	for _, v := range pods.Items {
+		// only pods that are actually running should be considered
+		if v.Status.Phase == v1.PodRunning {
+			for _, condition := range v.Status.Conditions {
+				if condition.Status != v1.ConditionTrue {
+					continue items
+				}
+			}
+
+			if group, have := v.Labels[podAffinityGroupLabel]; have {
+				log.Debugf("Namespace: %s, PodName: %s, PodIP: %s, Host: %s\n", v.Namespace, v.Name, v.Status.PodIP, v.Spec.NodeName)
+				rwPods = append(rwPods, &volthaPod{
+					name:    v.Name,
+					ipAddr:  v.Status.PodIP,
+					node:    v.Spec.NodeName,
+					devIds:  make(map[string]struct{}),
+					backend: afrouterRWClusterName + group,
+				})
+			} else {
+				log.Warnf("Pod %s found matching % without label %", v.Name, podLabelSelector, podAffinityGroupLabel)
+			}
+		}
+	}
+	return rwPods, nil
+}
+
+func reconcilePodDeviceIds(ctx context.Context, pod *volthaPod, ids map[string]struct{}) {
+	ctxTimeout, _ := context.WithTimeout(ctx, time.Second*5)
+	conn, err := Connect(ctxTimeout, fmt.Sprintf("%s:%d", pod.ipAddr, podGrpcPort))
+	if err != nil {
+		log.Debugf("Could not reconcile devices from %s, could not connect: %s", pod.name, err)
+		return
+	}
+	defer conn.Close()
+
+	var idList cmn.IDs
+	for k := range ids {
+		idList.Items = append(idList.Items, &cmn.ID{Id: k})
+	}
+
+	client := vpb.NewVolthaServiceClient(conn)
+	_, err = client.ReconcileDevices(ctx, &idList)
+	if err != nil {
+		log.Errorf("Attempt to reconcile ids on pod %s failed: %s", pod.name, err)
+		return
+	}
+}
+
+func queryPodDeviceIds(ctx context.Context, pod *volthaPod) map[string]struct{} {
+	ctxTimeout, _ := context.WithTimeout(ctx, time.Second*5)
+	conn, err := Connect(ctxTimeout, fmt.Sprintf("%s:%d", pod.ipAddr, podGrpcPort))
+	if err != nil {
+		log.Debugf("Could not query devices from %s, could not connect: %s", pod.name, err)
+		return nil
+	}
+	defer conn.Close()
+
+	client := vpb.NewVolthaServiceClient(conn)
+	devs, err := client.ListDeviceIds(ctx, &empty.Empty{})
+	if err != nil {
+		log.Error(err)
+		return nil
+	}
+
+	var ret = make(map[string]struct{})
+	for _, dv := range devs.Items {
+		ret[dv.Id] = struct{}{}
+	}
+	return ret
+}
+
+// coreMonitor polls the list of devices from all RW cores, pushes these devices
+// into the affinity router, and ensures that all cores in a backend have their devices synced
+func CoreMonitor(ctx context.Context, client pb.ConfigurationClient, clientset *kubernetes.Clientset) {
+	// map[backend]map[deviceId]struct{}
+	deviceOwnership := make(map[string]map[string]struct{})
+loop:
+	for {
+		// get the rw core list from k8s
+		rwPods, err := getVolthaPods(clientset)
+		if err != nil {
+			log.Error(err)
+			continue
+		}
+
+		// for every pod
+		for _, pod := range rwPods {
+			// get the devices for this pod's backend
+			devices, have := deviceOwnership[pod.backend]
+			if !have {
+				devices = make(map[string]struct{})
+				deviceOwnership[pod.backend] = devices
+			}
+
+			coreDevices := queryPodDeviceIds(ctx, pod)
+
+			// handle devices that exist in the core, but we have just learned about
+			for deviceId := range coreDevices {
+				// if there's a new device
+				if _, have := devices[deviceId]; !have {
+					// add the device to our local list
+					devices[deviceId] = struct{}{}
+					// push the device into the affinity router
+					setAffinity(ctx, client, deviceId, pod.backend)
+				}
+			}
+
+			// ensure that the core knows about all devices in its backend
+			toSync := make(map[string]struct{})
+			for deviceId := range devices {
+				// if the pod is missing any devices
+				if _, have := coreDevices[deviceId]; !have {
+					// we will reconcile them
+					toSync[deviceId] = struct{}{}
+				}
+			}
+
+			if len(toSync) != 0 {
+				reconcilePodDeviceIds(ctx, pod, toSync)
+			}
+		}
+
+		select {
+		case <-ctx.Done():
+			// if we're done, exit
+			break loop
+		case <-time.After(10 * time.Second): // wait a while
+		}
+	}
+}
diff --git a/internal/pkg/afrouterd/discoveryMonitor.go b/internal/pkg/afrouterd/discoveryMonitor.go
new file mode 100644
index 0000000..2fab6ed
--- /dev/null
+++ b/internal/pkg/afrouterd/discoveryMonitor.go
@@ -0,0 +1,112 @@
+/*
+ * 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 afrouterd
+
+import (
+	"errors"
+	"github.com/golang/protobuf/ptypes"
+	"github.com/opencord/voltha-go/common/log"
+	"github.com/opencord/voltha-go/kafka"
+	pb "github.com/opencord/voltha-protos/go/afrouter"
+	ic "github.com/opencord/voltha-protos/go/inter_container"
+	"golang.org/x/net/context"
+	"regexp"
+	"time"
+)
+
+func newKafkaClient(clientType string, host string, port int, instanceID string) (kafka.Client, error) {
+	log.Infow("kafka-client-type", log.Fields{"client": clientType})
+	switch clientType {
+	case "sarama":
+		return kafka.NewSaramaClient(
+			kafka.Host(host),
+			kafka.Port(port),
+			kafka.ConsumerType(kafka.GroupCustomer),
+			kafka.ProducerReturnOnErrors(true),
+			kafka.ProducerReturnOnSuccess(true),
+			kafka.ProducerMaxRetries(6),
+			kafka.NumPartitions(3),
+			kafka.ConsumerGroupName(instanceID),
+			kafka.ConsumerGroupPrefix(instanceID),
+			kafka.AutoCreateTopic(false),
+			kafka.ProducerFlushFrequency(5),
+			kafka.ProducerRetryBackoff(time.Millisecond*30)), nil
+	}
+	return nil, errors.New("unsupported-client-type")
+}
+
+func monitorDiscovery(kc kafka.Client, ctx context.Context, client pb.ConfigurationClient, ch <-chan *ic.InterContainerMessage, doneCh chan<- struct{}) {
+	defer close(doneCh)
+	defer kc.Stop()
+
+monitorLoop:
+	for {
+		select {
+		case <-ctx.Done():
+			break monitorLoop
+		case msg := <-ch:
+			log.Debug("Received a device discovery notification")
+			device := &ic.DeviceDiscovered{}
+			if err := ptypes.UnmarshalAny(msg.Body, device); err != nil {
+				log.Errorf("Could not unmarshal received notification %v", msg)
+			} else {
+				// somewhat hackish solution, backend is known from the first digit found in the publisher name
+				group := regexp.MustCompile(`\d`).FindString(device.Publisher)
+				if group != "" {
+					// set the affinity of the discovered device
+					setAffinity(ctx, client, device.Id, afrouterRWClusterName+group)
+				} else {
+					log.Error("backend is unknown")
+				}
+			}
+		}
+	}
+}
+
+func StartDiscoveryMonitor(ctx context.Context, client pb.ConfigurationClient) (<-chan struct{}, error) {
+	doneCh := make(chan struct{})
+	// Connect to kafka for discovery events
+	kc, err := newKafkaClient(kafkaClientType, kafkaHost, kafkaPort, kafkaInstanceID)
+	if err != nil {
+		panic(err)
+	}
+
+	for {
+		if err := kc.Start(); err != nil {
+			log.Error("Could not connect to kafka")
+		} else {
+			break
+		}
+		select {
+		case <-ctx.Done():
+			close(doneCh)
+			return doneCh, errors.New("GRPC context done")
+
+		case <-time.After(5 * time.Second):
+		}
+	}
+	ch, err := kc.Subscribe(&kafka.Topic{Name: kafkaTopic})
+	if err != nil {
+		log.Errorf("Could not subscribe to the '%s' channel, discovery disabled", kafkaTopic)
+		close(doneCh)
+		kc.Stop()
+		return doneCh, err
+	}
+
+	go monitorDiscovery(kc, ctx, client, ch, doneCh)
+	return doneCh, nil
+}
diff --git a/internal/pkg/afrouterd/k8s.go b/internal/pkg/afrouterd/k8s.go
new file mode 100644
index 0000000..16ed7f3
--- /dev/null
+++ b/internal/pkg/afrouterd/k8s.go
@@ -0,0 +1,51 @@
+/*
+ * 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 afrouterd
+
+import (
+	"github.com/opencord/voltha-go/common/log"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+)
+
+func K8sClientSet() *kubernetes.Clientset {
+	var config *rest.Config
+	if k8sApiServer != "" || k8sKubeConfigPath != "" {
+		// use combination of URL & local kube-config file
+		c, err := clientcmd.BuildConfigFromFlags(k8sApiServer, k8sKubeConfigPath)
+		if err != nil {
+			panic(err)
+		}
+		config = c
+	} else {
+		// use in-cluster config
+		c, err := rest.InClusterConfig()
+		if err != nil {
+			log.Errorf("Unable to load in-cluster config.  Try setting K8S_API_SERVER and K8S_KUBE_CONFIG_PATH?")
+			panic(err)
+		}
+		config = c
+	}
+	// creates the clientset
+	clientset, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		panic(err)
+	}
+
+	return clientset
+}
diff --git a/internal/pkg/afrouterd/misc.go b/internal/pkg/afrouterd/misc.go
new file mode 100644
index 0000000..09f3a8f
--- /dev/null
+++ b/internal/pkg/afrouterd/misc.go
@@ -0,0 +1,125 @@
+/*
+ * 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 afrouterd
+
+import (
+	"fmt"
+	"github.com/opencord/voltha-go/common/log"
+	pb "github.com/opencord/voltha-protos/go/afrouter"
+	"golang.org/x/net/context"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/connectivity"
+	"google.golang.org/grpc/keepalive"
+	"math"
+	"os"
+	"strconv"
+	"time"
+)
+
+type volthaPod struct {
+	name    string
+	ipAddr  string
+	node    string
+	devIds  map[string]struct{}
+	backend string
+}
+
+// TODO: These variables should be passed in from main() rather than
+// declared here.
+
+var (
+	// if k8s variables are undefined, will attempt to use in-cluster config
+	k8sApiServer      = GetStrEnv("K8S_API_SERVER", "")
+	k8sKubeConfigPath = GetStrEnv("K8S_KUBE_CONFIG_PATH", "")
+
+	podNamespace          = GetStrEnv("POD_NAMESPACE", "voltha")
+	podLabelSelector      = GetStrEnv("POD_LABEL_SELECTOR", "app=rw-core")
+	podAffinityGroupLabel = GetStrEnv("POD_AFFINITY_GROUP_LABEL", "affinity-group")
+
+	podGrpcPort = uint64(GetIntEnv("POD_GRPC_PORT", 0, math.MaxUint16, 50057))
+
+	afrouterRouterName    = GetStrEnv("AFROUTER_ROUTER_NAME", "vcore")
+	afrouterRouteName     = GetStrEnv("AFROUTER_ROUTE_NAME", "dev_manager")
+	afrouterRWClusterName = GetStrEnv("AFROUTER_RW_CLUSTER_NAME", "vcore")
+
+	kafkaTopic      = GetStrEnv("KAFKA_TOPIC", "AffinityRouter")
+	kafkaClientType = GetStrEnv("KAFKA_CLIENT_TYPE", "sarama")
+	kafkaHost       = GetStrEnv("KAFKA_HOST", "kafka")
+	kafkaPort       = GetIntEnv("KAFKA_PORT", 0, math.MaxUint16, 9092)
+	kafkaInstanceID = GetStrEnv("KAFKA_INSTANCE_ID", "arouterd")
+)
+
+func GetIntEnv(key string, min, max, defaultValue int) int {
+	if val, have := os.LookupEnv(key); have {
+		num, err := strconv.Atoi(val)
+		if err != nil || !(min <= num && num <= max) {
+			panic(fmt.Errorf("%s must be a number in the range [%d, %d]; default: %d", key, min, max, defaultValue))
+		}
+		return num
+	}
+	return defaultValue
+}
+
+func GetStrEnv(key, defaultValue string) string {
+	if val, have := os.LookupEnv(key); have {
+		return val
+	}
+	return defaultValue
+}
+
+func Connect(ctx context.Context, addr string) (*grpc.ClientConn, error) {
+	log.Debugf("Trying to connect to %s", addr)
+	conn, err := grpc.DialContext(ctx, addr,
+		grpc.WithInsecure(),
+		grpc.WithBlock(),
+		grpc.WithBackoffMaxDelay(time.Second*5),
+		grpc.WithKeepaliveParams(keepalive.ClientParameters{Time: time.Second * 10, Timeout: time.Second * 5}))
+	if err == nil {
+		log.Debugf("Connection succeeded")
+	}
+	return conn, err
+}
+
+func setAffinity(ctx context.Context, client pb.ConfigurationClient, deviceId string, backend string) {
+	log.Debugf("Configuring backend %s with device id %s \n", backend, deviceId)
+	if res, err := client.SetAffinity(ctx, &pb.Affinity{
+		Router:  afrouterRouterName,
+		Route:   afrouterRouteName,
+		Cluster: afrouterRWClusterName,
+		Backend: backend,
+		Id:      deviceId,
+	}); err != nil {
+		log.Debugf("failed affinity RPC call: %s\n", err)
+	} else {
+		log.Debugf("Result: %v\n", res)
+	}
+}
+
+// endOnClose cancels the context when the connection closes
+func ConnectionActiveContext(conn *grpc.ClientConn) context.Context {
+	ctx, disconnected := context.WithCancel(context.Background())
+	go func() {
+		for state := conn.GetState(); state != connectivity.TransientFailure && state != connectivity.Shutdown; state = conn.GetState() {
+			if !conn.WaitForStateChange(context.Background(), state) {
+				break
+			}
+		}
+		log.Infof("Connection to afrouter lost")
+		disconnected()
+	}()
+	return ctx
+}
diff --git a/internal/pkg/afrouterd/misc_test.go b/internal/pkg/afrouterd/misc_test.go
new file mode 100644
index 0000000..a07b9aa
--- /dev/null
+++ b/internal/pkg/afrouterd/misc_test.go
@@ -0,0 +1,100 @@
+/*
+ * 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.
+ */
+
+// This file implements an exit handler that tries to shut down all the
+// running servers before finally exiting. There are 2 triggers to this
+// clean exit thread: signals and an exit channel.
+
+package afrouterd
+
+import (
+	"github.com/stretchr/testify/assert"
+	"os"
+	"testing"
+)
+
+func TestGetIntEnv(t *testing.T) {
+
+	err := os.Setenv("testkey", "123")
+	assert.Nil(t, err)
+
+	defer func() { os.Unsetenv("testkey") }()
+
+	v := GetIntEnv("testkey", 0, 1000, 456)
+	assert.Equal(t, 123, v)
+
+	v = GetIntEnv("doesnotexist", 0, 1000, 456)
+	assert.Equal(t, 456, v)
+}
+
+func TestGetIntEnvTooLow(t *testing.T) {
+
+	err := os.Setenv("testkey", "-1")
+	assert.Nil(t, err)
+
+	defer func() { os.Unsetenv("testkey") }()
+
+	defer func() {
+		if r := recover(); r == nil {
+			t.Errorf("The code did not panic")
+		}
+	}()
+	_ = GetIntEnv("testkey", 0, 1000, 456)
+}
+
+func TestGetIntEnvTooHigh(t *testing.T) {
+
+	err := os.Setenv("testkey", "1001")
+	assert.Nil(t, err)
+
+	defer func() { os.Unsetenv("testkey") }()
+
+	defer func() {
+		if r := recover(); r == nil {
+			t.Errorf("The code did not panic")
+		}
+	}()
+	_ = GetIntEnv("testkey", 0, 1000, 456)
+}
+
+func TestGetIntEnvNotInteger(t *testing.T) {
+
+	err := os.Setenv("testkey", "stuff")
+	assert.Nil(t, err)
+
+	defer func() { os.Unsetenv("testkey") }()
+
+	defer func() {
+		if r := recover(); r == nil {
+			t.Errorf("The code did not panic")
+		}
+	}()
+	_ = GetIntEnv("testkey", 0, 1000, 456)
+}
+
+func TestGetStrEnv(t *testing.T) {
+
+	err := os.Setenv("testkey", "abc")
+	assert.Nil(t, err)
+
+	defer func() { os.Unsetenv("testkey") }()
+
+	v := GetStrEnv("testkey", "def")
+	assert.Equal(t, "abc", v)
+
+	v = GetStrEnv("doesnotexist", "def")
+	assert.Equal(t, "def", v)
+}