Clear UP4 state after forwarding config is retrieved and close P4rt stream on channel setup error

pfcpiface/p4rt_translator.go

@@ -54,10 +54,10 @@ type tunnelParams struct {
 }
 
 type P4rtTranslator struct {
-	p4Info p4ConfigV1.P4Info
+	p4Info *p4ConfigV1.P4Info
 }
 
-func newP4RtTranslator(p4info p4ConfigV1.P4Info) *P4rtTranslator {
+func newP4RtTranslator(p4info *p4ConfigV1.P4Info) *P4rtTranslator {
 	return &P4rtTranslator{
 		p4Info: p4info,
 	}

pfcpiface/p4rtc.go

@@ -51,7 +51,7 @@ type P4rtClient struct {
 	digests    chan *p4.DigestList
 
 	// exported fields
-	P4Info p4ConfigV1.P4Info
+	P4Info *p4ConfigV1.P4Info
 }
 
 type P4RuntimeError struct {
@@ -446,7 +446,7 @@ func (c *P4rtClient) GetForwardingPipelineConfig() (err error) {
 			"Operation successful, but no P4 config provided.")
 	}
 
-	c.P4Info = *pipeline.Config.P4Info
+	c.P4Info = pipeline.Config.P4Info
 
 	getLog.Info("Got ForwardingPipelineConfig from P4Rt device")
 
@@ -479,15 +479,15 @@ func (c *P4rtClient) SetForwardingPipelineConfig(p4InfoPath, deviceConfigPath st
 		return
 	}
 
-	var p4info p4ConfigV1.P4Info
+	p4Info := &p4ConfigV1.P4Info{}
 
-	err = proto.UnmarshalText(string(p4infoBytes), &p4info)
+	err = proto.UnmarshalText(string(p4infoBytes), p4Info)
 	if err != nil {
 		log.Println("Unmarshal test failed for p4info ", err)
 		return
 	}
 
-	c.P4Info = p4info
+	c.P4Info = p4Info
 
 	deviceConfig, err := LoadDeviceConfig(deviceConfigPath)
 	if err != nil {
@@ -496,7 +496,7 @@ func (c *P4rtClient) SetForwardingPipelineConfig(p4InfoPath, deviceConfigPath st
 	}
 
 	var pipeline p4.ForwardingPipelineConfig
-	pipeline.P4Info = &p4info
+	pipeline.P4Info = p4Info
 	pipeline.P4DeviceConfig = deviceConfig
 
 	err = SetPipelineConfig(c.client, c.deviceID, &c.electionID, &pipeline)
@@ -588,9 +588,26 @@ func CreateChannel(host string, deviceID uint64) (*P4rtClient, error) {
 		return nil, err
 	}
 
+	closeStreamOnError := func() {
+		if client.stream != nil {
+			err := client.stream.CloseSend()
+			if err != nil {
+				log.Errorf("Failed to close P4Rt stream with %v: %v", client.conn.Target(), err)
+			}
+		}
+	}
+
 	err = client.SetMastership(TimeBasedElectionId())
 	if err != nil {
 		log.Error("Set Mastership error: ", err)
+		closeStreamOnError()
+
+		return nil, err
+	}
+
+	err = client.GetForwardingPipelineConfig()
+	if err != nil {
+		closeStreamOnError()
 		return nil, err
 	}
 

pfcpiface/up4.go

@@ -291,11 +291,6 @@ func (up4 *UP4) setupChannel() error {
 
 	up4.p4client = client
 
-	err = up4.p4client.GetForwardingPipelineConfig()
-	if err != nil {
-		return err
-	}
-
 	up4.p4RtTranslator = newP4RtTranslator(up4.p4client.P4Info)
 
 	setupLog.Debug("P4Rt channel created")
@@ -466,16 +461,16 @@ func (up4 *UP4) tryConnect() error {
 		return nil
 	}
 
-	// p4client is nil only if it's a first attempt to connect to UP4.
-	firstRun := up4.p4client == nil
+	// datapath state should be cleared & initialized if P4Rt connection or ForwardingConfig is not setup yet.
+	shouldClearAndInitialize := up4.p4client == nil || up4.p4client.P4Info == nil
 
 	err := up4.setupChannel()
 	if err != nil {
 		log.Errorf("Failed to setup UP4 channel: %v", err)
 		return err
 	}
 
-	err = up4.initialize(firstRun)
+	err = up4.initialize(shouldClearAndInitialize)
 	if err != nil {
 		log.Errorf("Failed to initialize UP4: %v", err)
 		return err
@@ -585,10 +580,10 @@ func (up4 *UP4) clearDatapathState() error {
 
 // initialize configures the UP4-related objects.
 // A caller should ensure that P4Client is not nil and the P4Runtime channel is open.
-func (up4 *UP4) initialize(firstRun bool) error {
+func (up4 *UP4) initialize(shouldClear bool) error {
 	// always clear datapath state at startup or
 	// on UP4 datapath restart if ClearStateOnRestart is enabled.
-	if firstRun || up4.conf.ClearStateOnRestart {
+	if shouldClear || up4.conf.ClearStateOnRestart {
 		if err := up4.clearDatapathState(); err != nil {
 			return err
 		}