Advanced ODC information OCTRL-725

core/integration/odc/plugin.go

@@ -40,6 +40,7 @@ import (
 
 	"github.com/AliceO2Group/Control/apricot"
 	"github.com/AliceO2Group/Control/common/logger/infologger"
+	"github.com/AliceO2Group/Control/common/utils"
 	"github.com/AliceO2Group/Control/common/utils/uid"
 	"github.com/AliceO2Group/Control/core/integration"
 	odc "github.com/AliceO2Group/Control/core/integration/odc/protos"
@@ -81,8 +82,21 @@ type OdcStatus struct {
 }
 
 type OdcPartitionInfo struct {
-	RunNumber uint32
-	State     string
+	PartitionId      uid.ID      `json:"-"`
+	RunNumber        uint32      `json:"runNumber"`
+	State            string      `json:"state"`
+	DdsSessionId     string      `json:"ddsSessionId"`
+	DdsSessionStatus string      `json:"ddsSessionStatus"`
+	Devices          []OdcDevice `json:"devices"`
+	Hosts            []string    `json:"hosts"`
+}
+
+type OdcDevice struct {
+	TaskId  string `json:"taskId"`
+	State   string `json:"state"`
+	Path    string `json:"path"`
+	Ignored bool   `json:"ignored"`
+	Host    string `json:"host"`
 }
 
 func NewPlugin(endpoint string) integration.Plugin {
@@ -123,6 +137,7 @@ func (p *Plugin) GetConnectionState() string {
 }
 
 func (p *Plugin) queryPartitionStatus() {
+	defer utils.TimeTrackFunction(time.Now(), log.WithPrefix("odcclient"))
 	ctx, cancel := context.WithTimeout(context.Background(), ODC_STATUS_TIMEOUT)
 	defer cancel()
 
@@ -148,16 +163,79 @@ func (p *Plugin) queryPartitionStatus() {
 		Error:      statusRep.Error,
 		Partitions: make(map[uid.ID]OdcPartitionInfo),
 	}
-	for _, v := range statusRep.Partitions {
+
+	odcPartInfoSlice := make([]*OdcPartitionInfo, len(statusRep.Partitions))
+
+	// concurrent request for the detailed state of each partition
+	var wg sync.WaitGroup
+
+	for idx, odcPartSt := range statusRep.Partitions {
+		if odcPartSt == nil {
+			continue
+		}
 		var id uid.ID
-		id, err = uid.FromString(v.Partitionid)
+		id, err = uid.FromString(odcPartSt.Partitionid)
 		if err != nil {
 			continue
 		}
-		response.Partitions[id] = OdcPartitionInfo{
-			RunNumber: uint32(v.Runnr),
-			State:     v.State,
+
+		wg.Add(1)
+
+		odcPartInfoSlice[idx] = &OdcPartitionInfo{
+			PartitionId:      id,
+			RunNumber:        uint32(odcPartSt.Runnr),
+			State:            odcPartSt.State,
+			DdsSessionId:     odcPartSt.Sessionid,
+			DdsSessionStatus: odcPartSt.Status.String(),
+		}
+		i := idx
+
+		go func(idx int, partId uid.ID) {
+			defer wg.Done()
+
+			ctx, cancel := context.WithTimeout(context.Background(), ODC_STATUS_TIMEOUT)
+			defer cancel()
+
+			odcPartStateRep, err := p.odcClient.GetState(ctx, &odc.StateRequest{
+				Partitionid: partId.String(),
+				Detailed:    true,
+			}, grpc.EmptyCallOption{})
+			if err != nil {
+				log.WithField("level", infologger.IL_Support).
+					WithField("call", "GetState").
+					WithField("partition", partId.String()).
+					WithError(err).Error("ODC error")
+				return
+			}
+			if odcPartStateRep == nil || odcPartStateRep.Reply == nil {
+				log.WithField("level", infologger.IL_Support).
+					WithField("call", "GetState").
+					WithField("partition", partId.String()).
+					WithError(fmt.Errorf("ODC GetState response is nil")).Error("ODC error")
+				return
+			}
+
+			odcPartInfoSlice[idx].Hosts = odcPartStateRep.Reply.Hosts
+			odcPartInfoSlice[idx].Devices = make([]OdcDevice, len(odcPartStateRep.Devices))
+			for i, device := range odcPartStateRep.Devices {
+				odcPartInfoSlice[idx].Devices[i] = OdcDevice{
+					TaskId:  strconv.FormatUint(device.Id, 10),
+					State:   device.State,
+					Path:    device.Path,
+					Ignored: device.Ignored,
+					Host:    device.Host,
+				}
+			}
+		}(i, id)
+	}
+	wg.Wait()
+
+	for _, odcPartSt := range odcPartInfoSlice {
+		if odcPartSt.PartitionId.IsNil() {
+			// The partition wasn't found in the ODC response
+			continue
 		}
+		response.Partitions[odcPartSt.PartitionId] = *odcPartSt
 	}
 
 	p.cachedStatusMu.Lock()