diff --git a/internal/pkg/core/device_handler.go b/internal/pkg/core/device_handler.go
index fa8dd52..ef1b179 100644
--- a/internal/pkg/core/device_handler.go
+++ b/internal/pkg/core/device_handler.go
@@ -45,6 +45,7 @@
 	"github.com/opencord/voltha-openolt-adapter/internal/pkg/olterrors"
 	rsrcMgr "github.com/opencord/voltha-openolt-adapter/internal/pkg/resourcemanager"
 	"github.com/opencord/voltha-protos/v4/go/common"
+	"github.com/opencord/voltha-protos/v4/go/extension"
 	ic "github.com/opencord/voltha-protos/v4/go/inter_container"
 	of "github.com/opencord/voltha-protos/v4/go/openflow_13"
 	oop "github.com/opencord/voltha-protos/v4/go/openolt"
@@ -62,6 +63,7 @@
 	McastFlowOrGroupAdd    = "McastFlowOrGroupAdd"
 	McastFlowOrGroupModify = "McastFlowOrGroupModify"
 	McastFlowOrGroupRemove = "McastFlowOrGroupRemove"
+	oltPortInfoTimeout     = 3
 )
 
 //DeviceHandler will interact with the OLT device.
@@ -2479,3 +2481,79 @@
 		}
 	}
 }
+
+func (dh *DeviceHandler) getOltPortCounters(ctx context.Context, oltPortInfo *extension.GetOltPortCounters) *extension.SingleGetValueResponse {
+
+	singleValResp := extension.SingleGetValueResponse{
+		Response: &extension.GetValueResponse{
+			Response: &extension.GetValueResponse_PortCoutners{
+				PortCoutners: &extension.GetOltPortCountersResponse{},
+			},
+		},
+	}
+
+	errResp := func(status extension.GetValueResponse_Status,
+		reason extension.GetValueResponse_ErrorReason) *extension.SingleGetValueResponse {
+		return &extension.SingleGetValueResponse{
+			Response: &extension.GetValueResponse{
+				Status:    status,
+				ErrReason: reason,
+			},
+		}
+	}
+
+	if oltPortInfo.PortType != extension.GetOltPortCounters_Port_ETHERNET_NNI &&
+		oltPortInfo.PortType != extension.GetOltPortCounters_Port_PON_OLT {
+		//send error response
+		logger.Debugw(ctx, "getOltPortCounters invalid portType", log.Fields{"oltPortInfo": oltPortInfo.PortType})
+		return errResp(extension.GetValueResponse_ERROR, extension.GetValueResponse_INVALID_PORT_TYPE)
+	}
+	statIndChn := make(chan bool, 1)
+	dh.portStats.RegisterForStatIndication(ctx, portStatsType, statIndChn, oltPortInfo.PortNo, oltPortInfo.PortType)
+	defer dh.portStats.DeRegisterFromStatIndication(ctx, portStatsType, statIndChn)
+	//request openOlt agent to send the the port statistics indication
+
+	go func() {
+		_, err := dh.Client.CollectStatistics(ctx, new(oop.Empty))
+		if err != nil {
+			logger.Errorw(ctx, "getOltPortCounters CollectStatistics failed ", log.Fields{"err": err})
+		}
+	}()
+	select {
+	case <-statIndChn:
+		//indication received for ports stats
+		logger.Debugw(ctx, "getOltPortCounters recvd statIndChn", log.Fields{"oltPortInfo": oltPortInfo})
+	case <-time.After(oltPortInfoTimeout * time.Second):
+		logger.Debugw(ctx, "getOltPortCounters timeout happened", log.Fields{"oltPortInfo": oltPortInfo})
+		return errResp(extension.GetValueResponse_ERROR, extension.GetValueResponse_TIMEOUT)
+	case <-ctx.Done():
+		logger.Debugw(ctx, "getOltPortCounters ctx Done ", log.Fields{"oltPortInfo": oltPortInfo})
+		return errResp(extension.GetValueResponse_ERROR, extension.GetValueResponse_TIMEOUT)
+	}
+	if oltPortInfo.PortType == extension.GetOltPortCounters_Port_ETHERNET_NNI {
+		//get nni stats
+		intfID := PortNoToIntfID(oltPortInfo.PortNo, voltha.Port_ETHERNET_NNI)
+		logger.Debugw(ctx, "getOltPortCounters intfID  ", log.Fields{"intfID": intfID})
+		cmnni := dh.portStats.collectNNIMetrics(intfID)
+		if cmnni == nil {
+			//TODO define the error reason
+			return errResp(extension.GetValueResponse_ERROR, extension.GetValueResponse_INTERNAL_ERROR)
+		}
+		dh.portStats.updateGetOltPortCountersResponse(ctx, &singleValResp, cmnni)
+		return &singleValResp
+
+	} else if oltPortInfo.PortType == extension.GetOltPortCounters_Port_PON_OLT {
+		// get pon stats
+		intfID := PortNoToIntfID(oltPortInfo.PortNo, voltha.Port_PON_OLT)
+		if val, ok := dh.activePorts.Load(intfID); ok && val == true {
+			cmpon := dh.portStats.collectPONMetrics(intfID)
+			if cmpon == nil {
+				//TODO define the error reason
+				return errResp(extension.GetValueResponse_ERROR, extension.GetValueResponse_INTERNAL_ERROR)
+			}
+			dh.portStats.updateGetOltPortCountersResponse(ctx, &singleValResp, cmpon)
+			return &singleValResp
+		}
+	}
+	return errResp(extension.GetValueResponse_ERROR, extension.GetValueResponse_INTERNAL_ERROR)
+}
diff --git a/internal/pkg/core/openolt.go b/internal/pkg/core/openolt.go
index 49c5675..eebed78 100644
--- a/internal/pkg/core/openolt.go
+++ b/internal/pkg/core/openolt.go
@@ -29,6 +29,7 @@
 	"github.com/opencord/voltha-lib-go/v4/pkg/log"
 	"github.com/opencord/voltha-openolt-adapter/internal/pkg/config"
 	"github.com/opencord/voltha-openolt-adapter/internal/pkg/olterrors"
+	"github.com/opencord/voltha-protos/v4/go/extension"
 	ic "github.com/opencord/voltha-protos/v4/go/inter_container"
 	"github.com/opencord/voltha-protos/v4/go/openflow_13"
 	"github.com/opencord/voltha-protos/v4/go/voltha"
@@ -376,3 +377,29 @@
 	}
 	return resp, nil
 }
+
+//Single_get_value_request handles get uni status on ONU and ondemand metric on OLT
+func (oo *OpenOLT) Single_get_value_request(ctx context.Context, request extension.SingleGetValueRequest) (*extension.SingleGetValueResponse, error) {
+	logger.Infow(ctx, "Single_get_value_request", log.Fields{"request": request})
+
+	errResp := func(status extension.GetValueResponse_Status,
+		reason extension.GetValueResponse_ErrorReason) *extension.SingleGetValueResponse {
+		return &extension.SingleGetValueResponse{
+			Response: &extension.GetValueResponse{
+				Status:    status,
+				ErrReason: reason,
+			},
+		}
+	}
+	if handler := oo.getDeviceHandler(request.TargetId); handler != nil {
+		switch reqType := request.GetRequest().GetRequest().(type) {
+		case *extension.GetValueRequest_OltPortInfo:
+			return handler.getOltPortCounters(ctx, reqType.OltPortInfo), nil
+		default:
+			return errResp(extension.GetValueResponse_ERROR, extension.GetValueResponse_UNSUPPORTED), nil
+		}
+	}
+
+	logger.Infow(ctx, "Single_get_value_request failed ", log.Fields{"request": request})
+	return errResp(extension.GetValueResponse_ERROR, extension.GetValueResponse_INVALID_DEVICE_ID), nil
+}
diff --git a/internal/pkg/core/statsmanager.go b/internal/pkg/core/statsmanager.go
index 1b52e7c..e13abed 100755
--- a/internal/pkg/core/statsmanager.go
+++ b/internal/pkg/core/statsmanager.go
@@ -18,6 +18,7 @@
 package core
 
 import (
+	"container/list"
 	"context"
 	"fmt"
 	rsrcMgr "github.com/opencord/voltha-openolt-adapter/internal/pkg/resourcemanager"
@@ -26,6 +27,7 @@
 
 	"github.com/opencord/voltha-lib-go/v4/pkg/log"
 	"github.com/opencord/voltha-openolt-adapter/internal/pkg/olterrors"
+	"github.com/opencord/voltha-protos/v4/go/extension"
 	"github.com/opencord/voltha-protos/v4/go/openolt"
 	"github.com/opencord/voltha-protos/v4/go/voltha"
 )
@@ -93,6 +95,14 @@
 var onuStats = make(chan *openolt.OnuStatistics, 100)
 var gemStats = make(chan *openolt.GemPortStatistics, 100)
 
+//statRegInfo is used to register for notifications
+//on receiving port stats and flow stats indication
+type statRegInfo struct {
+	chn      chan bool
+	portNo   uint32
+	portType extension.GetOltPortCounters_PortType
+}
+
 // PonPort representation
 type PonPort struct {
 	/*
@@ -242,12 +252,23 @@
 	return &NNI
 }
 
+//StatType defines portStatsType and flowStatsType types
+type StatType int
+
+const (
+	portStatsType StatType = iota
+	flowStatsType
+)
+
 // OpenOltStatisticsMgr structure
 type OpenOltStatisticsMgr struct {
 	Device         *DeviceHandler
 	NorthBoundPort map[uint32]*NniPort
 	SouthBoundPort map[uint32]*PonPort
 	// TODO  PMMetrics Metrics
+	//statIndListners is the list of requests to be notified when port and flow stats indication is received
+	statIndListnerMu sync.Mutex
+	statIndListners  map[StatType]*list.List
 }
 
 // NewOpenOltStatsMgr returns a new instance of the OpenOltStatisticsMgr
@@ -271,6 +292,9 @@
 	if StatMgr.Device.openOLT.enableGemStats {
 		go StatMgr.publishGemStats()
 	}
+	StatMgr.statIndListners = make(map[StatType]*list.List)
+	StatMgr.statIndListners[portStatsType] = list.New()
+	StatMgr.statIndListners[flowStatsType] = list.New()
 	return &StatMgr
 }
 
@@ -529,7 +553,7 @@
 }
 
 // publishMetrics will publish the pon port metrics
-func (StatMgr OpenOltStatisticsMgr) publishMetrics(ctx context.Context, statType string, val map[string]float32,
+func (StatMgr *OpenOltStatisticsMgr) publishMetrics(ctx context.Context, statType string, val map[string]float32,
 	port *voltha.Port, devID string, devType string) {
 	logger.Debugw(ctx, "publish-metrics",
 		log.Fields{
@@ -576,6 +600,9 @@
 func (StatMgr *OpenOltStatisticsMgr) PortStatisticsIndication(ctx context.Context, PortStats *openolt.PortStatistics, NumPonPorts uint32) {
 	StatMgr.PortsStatisticsKpis(ctx, PortStats, NumPonPorts)
 	logger.Debugw(ctx, "received-port-stats-indication", log.Fields{"port-stats": PortStats})
+	//Indicate that PortStatisticsIndication is handled
+	//PortStats.IntfId is actually the port number
+	StatMgr.processStatIndication(ctx, portStatsType, PortStats.IntfId)
 	// TODO send stats to core topic to the voltha kafka or a different kafka ?
 }
 
@@ -673,3 +700,81 @@
 	*/
 
 }
+
+func (StatMgr *OpenOltStatisticsMgr) updateGetOltPortCountersResponse(ctx context.Context, singleValResp *extension.SingleGetValueResponse, stats map[string]float32) {
+
+	metrics := singleValResp.GetResponse().GetPortCoutners()
+	metrics.TxBytes = uint64(stats["TxBytes"])
+	metrics.RxBytes = uint64(stats["RxBytes"])
+	metrics.TxPackets = uint64(stats["TxPackets"])
+	metrics.RxPackets = uint64(stats["RxPackets"])
+	metrics.TxErrorPackets = uint64(stats["TxErrorPackets"])
+	metrics.RxErrorPackets = uint64(stats["RxErrorPackets"])
+	metrics.TxBcastPackets = uint64(stats["TxBcastPackets"])
+	metrics.RxBcastPackets = uint64(stats["RxBcastPackets"])
+	metrics.TxUcastPackets = uint64(stats["TxUcastPackets"])
+	metrics.RxUcastPackets = uint64(stats["RxUcastPackets"])
+	metrics.TxMcastPackets = uint64(stats["TxMcastPackets"])
+	metrics.RxMcastPackets = uint64(stats["RxMcastPackets"])
+
+	singleValResp.Response.Status = extension.GetValueResponse_OK
+	logger.Debugw(ctx, "updateGetOltPortCountersResponse", log.Fields{"resp": singleValResp})
+}
+
+//RegisterForStatIndication registers ch as a channel on which indication is sent when statistics of type t is received
+func (StatMgr *OpenOltStatisticsMgr) RegisterForStatIndication(ctx context.Context, t StatType, ch chan bool, portNo uint32, portType extension.GetOltPortCounters_PortType) {
+	statInd := statRegInfo{
+		chn:      ch,
+		portNo:   portNo,
+		portType: portType,
+	}
+
+	logger.Debugf(ctx, "RegisterForStatIndication stat type %v portno %v porttype %v chan %v", t, portNo, portType, ch)
+	StatMgr.statIndListnerMu.Lock()
+	StatMgr.statIndListners[t].PushBack(statInd)
+	StatMgr.statIndListnerMu.Unlock()
+
+}
+
+//DeRegisterFromStatIndication removes the previously registered channel ch for type t of statistics
+func (StatMgr *OpenOltStatisticsMgr) DeRegisterFromStatIndication(ctx context.Context, t StatType, ch chan bool) {
+	StatMgr.statIndListnerMu.Lock()
+	defer StatMgr.statIndListnerMu.Unlock()
+
+	for e := StatMgr.statIndListners[t].Front(); e != nil; e = e.Next() {
+		statInd := e.Value.(statRegInfo)
+		if statInd.chn == ch {
+			StatMgr.statIndListners[t].Remove(e)
+			return
+		}
+	}
+}
+
+func (StatMgr *OpenOltStatisticsMgr) processStatIndication(ctx context.Context, t StatType, portNo uint32) {
+	var deRegList []*list.Element
+	var statInd statRegInfo
+
+	StatMgr.statIndListnerMu.Lock()
+	defer StatMgr.statIndListnerMu.Unlock()
+
+	if StatMgr.statIndListners[t] == nil || StatMgr.statIndListners[t].Len() == 0 {
+		logger.Debugf(ctx, "processStatIndication %v list is empty ", t)
+		return
+	}
+
+	for e := StatMgr.statIndListners[t].Front(); e != nil; e = e.Next() {
+		statInd = e.Value.(statRegInfo)
+		if statInd.portNo != portNo {
+			fmt.Printf("Skipping %v\n", e.Value)
+			continue
+		}
+		// message sent
+		statInd.chn <- true
+		deRegList = append(deRegList, e)
+
+	}
+	for _, e := range deRegList {
+		StatMgr.statIndListners[t].Remove(e)
+	}
+
+}