Automated cherry pick of #275: fix: notify may lost event

go.mod

@@ -5,7 +5,7 @@ go 1.18
 require (
 	github.com/coreos/go-iptables v0.6.0
 	github.com/digitalocean/go-openvswitch v0.0.0-20190515160856-1141932ed5cf
-	github.com/fsnotify/fsnotify v1.4.9
+	github.com/fsnotify/fsnotify v1.5.4
 	github.com/golang/protobuf v1.5.2
 	github.com/mitchellh/go-homedir v1.1.0
 	github.com/pkg/errors v0.9.1
@@ -19,7 +19,7 @@ require (
 	google.golang.org/protobuf v1.27.1
 	yunion.io/x/jsonutils v1.0.1-0.20230613121553-0f3b41e2ef19
 	yunion.io/x/log v1.0.1-0.20230411060016-feb3f46ab361
-	yunion.io/x/onecloud v0.0.0-20231225063626-a1ab937ef72a
+	yunion.io/x/onecloud v0.0.0-20231231043024-486aa4edf48b
 	yunion.io/x/pkg v1.0.1-0.20231101105448-abef64cdc142
 )
 
@@ -110,7 +110,7 @@ require (
 	moul.io/http2curl/v2 v2.3.0 // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.0.1 // indirect
 	sigs.k8s.io/yaml v1.2.0 // indirect
-	yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20231222093659-799fff1d62a1 // indirect
+	yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20231229064730-739383d4356c // indirect
 	yunion.io/x/executor v0.0.0-20230705125604-c5ac3141db32 // indirect
 	yunion.io/x/s3cli v0.0.0-20190917004522-13ac36d8687e // indirect
 	yunion.io/x/sqlchemy v1.1.3-0.20231219103237-8264e6ead2fc // indirect

go.sum

@@ -111,8 +111,9 @@ github.com/evanphx/json-patch v4.9.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLi
 github.com/fatih/color v1.13.0 h1:8LOYc1KYPPmyKMuN8QV2DNRWNbLo6LZ0iLs8+mlH53w=
 github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
-github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
+github.com/fsnotify/fsnotify v1.5.4 h1:jRbGcIw6P2Meqdwuo0H1p6JVLbL5DHKAKlYndzMwVZI=
+github.com/fsnotify/fsnotify v1.5.4/go.mod h1:OVB6XrOHzAwXMpEM7uPOzcehqUV2UqJxmVXmkdnm1bU=
 github.com/gernest/wow v0.1.0/go.mod h1:dEPabJRi5BneI1Nev1VWo0ZlcTWibHWp43qxKms4elY=
 github.com/ghodss/yaml v0.0.0-20150909031657-73d445a93680/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/ghodss/yaml v1.0.0 h1:wQHKEahhL6wmXdzwWG11gIVCkOv05bNOh+Rxn0yngAk=
@@ -598,6 +599,7 @@ golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220128215802-99c3d69c2c27/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220319134239-a9b59b0215f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.9.0 h1:KS/R3tvhPqvJvwcKfnBHJwwthS11LRhmM5D59eEXa0s=
@@ -818,8 +820,8 @@ sigs.k8s.io/structured-merge-diff/v4 v4.0.1/go.mod h1:bJZC9H9iH24zzfZ/41RGcq60oK
 sigs.k8s.io/yaml v1.1.0/go.mod h1:UJmg0vDUVViEyp3mgSv9WPwZCDxu4rQW1olrI1uml+o=
 sigs.k8s.io/yaml v1.2.0 h1:kr/MCeFWJWTwyaHoR9c8EjH9OumOmoF9YGiZd7lFm/Q=
 sigs.k8s.io/yaml v1.2.0/go.mod h1:yfXDCHCao9+ENCvLSE62v9VSji2MKu5jeNfTrofGhJc=
-yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20231222093659-799fff1d62a1 h1:AT43JqLQ9x+Vyp+/2tYvmfo6UAibZdImBGwOqsqhBFU=
-yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20231222093659-799fff1d62a1/go.mod h1:aj1gR9PPb6eqqKOwvANe26CoZFY8ydmXy0fuvgKYXH0=
+yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20231229064730-739383d4356c h1:EH6bmVx8fNlkej3oWilVwqVbCev9KT7QpOd8cqS7oMw=
+yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20231229064730-739383d4356c/go.mod h1:aj1gR9PPb6eqqKOwvANe26CoZFY8ydmXy0fuvgKYXH0=
 yunion.io/x/executor v0.0.0-20230705125604-c5ac3141db32 h1:v7POYkQwo1XzOxBoIoRVr/k0V9Y5JyjpshlIFa9raug=
 yunion.io/x/executor v0.0.0-20230705125604-c5ac3141db32/go.mod h1:Uxuou9WQIeJXNpy7t2fPLL0BYLvLiMvGQwY7Qc6aSws=
 yunion.io/x/jsonutils v0.0.0-20190625054549-a964e1e8a051/go.mod h1:4N0/RVzsYL3kH3WE/H1BjUQdFiWu50JGCFQuuy+Z634=
@@ -829,8 +831,8 @@ yunion.io/x/log v0.0.0-20190514041436-04ce53b17c6b/go.mod h1:+gauLs73omeJAPlsXce
 yunion.io/x/log v0.0.0-20190629062853-9f6483a7103d/go.mod h1:LC6f/4FozL0iaAbnFt2eDX9jlsyo3WiOUPm03d7+U4U=
 yunion.io/x/log v1.0.1-0.20230411060016-feb3f46ab361 h1:c5LyNdhbvBe/92pXs9jgXOU/S+RgYh/DHe538LpT/Mo=
 yunion.io/x/log v1.0.1-0.20230411060016-feb3f46ab361/go.mod h1:LC6f/4FozL0iaAbnFt2eDX9jlsyo3WiOUPm03d7+U4U=
-yunion.io/x/onecloud v0.0.0-20231225063626-a1ab937ef72a h1:+j/fhTN/zhkNRVk+PjHg9PsOphytOS2ldxsLavcuTXs=
-yunion.io/x/onecloud v0.0.0-20231225063626-a1ab937ef72a/go.mod h1:nfyNCrKew69Mg2feWRqI+9PIcFFSPVTL3q0/j2NDMjQ=
+yunion.io/x/onecloud v0.0.0-20231231043024-486aa4edf48b h1:YwhaQJRfXASQAWMpxXiE1nhIszmhtKL+0/c5tD4d6AA=
+yunion.io/x/onecloud v0.0.0-20231231043024-486aa4edf48b/go.mod h1:m6xaWD8pWU+WhwrsBAKzqu+uR6JKsWBkCk5A4177jCc=
 yunion.io/x/pkg v0.0.0-20190620104149-945c25821dbf/go.mod h1:t6rEGG2sQ4J7DhFxSZVOTjNd0YO/KlfWQyK1W4tog+E=
 yunion.io/x/pkg v0.0.0-20190628082551-f4033ba2ea30/go.mod h1:t6rEGG2sQ4J7DhFxSZVOTjNd0YO/KlfWQyK1W4tog+E=
 yunion.io/x/pkg v1.0.1-0.20231101105448-abef64cdc142 h1:L6LqxfP08eWUx+A6yQdrL6VB3KILU1TbQllPtKZq7PM=

pkg/agent/server/flowman.go

@@ -85,16 +85,19 @@ func (fm *FlowMan) doCheck() {
 	}
 	defer log.Infof("flowman %s: check done", fm.bridge)
 	var err error
+	// fs0: current flows
 	fs0, err := fm.doDumpFlows()
 	if err != nil {
+		log.Errorf("FlowMan doCheck doDumpFlows fail %s", err)
 		return
 	}
 	fsAdd := utils.NewFlowSet()
 	fsDel := utils.NewFlowSet()
-	flows1 := []*ovs.Flow{}
+	// flows1: flows to install
+	// flows1 := []*ovs.Flow{}
 	for _, fs1 := range fm.flowSets {
 		for _, f1 := range fs1.Flows() {
-			flows1 = append(flows1, f1)
+			// flows1 = append(flows1, f1)
 			if !fs0.Contains(f1) {
 				fsAdd.Add(f1)
 			}

pkg/agent/server/guest.go

@@ -249,7 +249,10 @@ func (g *Guest) UpdateSettings(ctx context.Context) {
 			g.watcher.agent.HostId(g.HostId)
 		}
 	case errNotRunning, errPortNotReady, errVolatileHost:
+		log.Debugf("guest %s(%s) ClearSettings due to g.refresh %s", g.Name, g.Id, err)
 		g.ClearSettings(ctx)
+	default:
+		log.Errorf("guest %s(%s) g.refresh error %s", g.Name, g.Id, err)
 	}
 }
 

pkg/agent/server/ifaces.go

@@ -136,6 +136,8 @@ func (ij *ifaceJanitor) scan(ctx context.Context) error {
 		}
 		wantMap.mergeWith(serversMap)
 	}
+	// log.Debugf("got Map: %s", jsonutils.Marshal(gotMap))
+	// log.Debugf("want Map: %s", jsonutils.Marshal(wantMap))
 	for br, ifaces := range gotMap {
 		for iface, _ := range ifaces {
 			if !wantMap.has(br, iface) {

pkg/agent/server/watch.go

@@ -16,6 +16,7 @@ package server
 
 import (
 	"context"
+	"fmt"
 	"io/ioutil"
 	"path"
 	"path/filepath"
@@ -100,12 +101,23 @@ const (
 	watchEventTypeDelServer
 )
 
+var watchEventTypeStringMap = []string{
+	"watchEventTypeAddServerDir",
+	"watchEventTypeDelServerDir",
+	"watchEventTypeUpdServer",
+	"watchEventTypeDelServer",
+}
+
 type watchEvent struct {
 	evType    watchEventType
 	guestId   string
 	guestPath string // path to the servers/<uuid> dir
 }
 
+func (w *watchEvent) String() string {
+	return fmt.Sprintf("type: %s guest_id: %s path: %s", watchEventTypeStringMap[w.evType], w.guestId, w.guestPath)
+}
+
 func (w *serversWatcher) scan(ctx context.Context) {
 	serversPath := w.hostConfig.ServersPath
 	fis, err := ioutil.ReadDir(serversPath)
@@ -132,6 +144,9 @@ func (w *serversWatcher) scan(ctx context.Context) {
 // addGuestWatch adds the server with <id> in <path> to watch list.  It returns
 // error when adding watch failed, but it will always return non-nil *Guest
 func (w *serversWatcher) addGuestWatch(id, path string) (*Guest, error) {
+	if g, ok := w.guests[id]; ok {
+		return g, nil
+	}
 	ug := &utils.Guest{
 		Id:         id,
 		Path:       path,
@@ -217,41 +232,46 @@ func (w *serversWatcher) Start(ctx context.Context, agent *AgentServer) {
 			pendingChan = pendingRefreshTicker.C
 		}
 		select {
-		case ev := <-w.watcher.Events:
+		case ev, ok := <-w.watcher.Events:
+			if !ok {
+				log.Errorf("fsnotity.watch.Events error")
+				goto out
+			}
 			wev := w.watchEvent(&ev)
 			if wev == nil {
 				log.Debugf("inotify event ignored: %s", ev)
-				break
-			}
-			guestId := wev.guestId
-			guestPath := wev.guestPath
-			switch wev.evType {
-			case watchEventTypeAddServerDir:
-				log.Infof("received guest path add event: %s", guestPath)
-				g, err := w.addGuestWatch(guestId, guestPath)
-				if err != nil {
-					log.Errorf("watch guest failed: %s: %s", guestPath, err)
-				}
-				g.UpdateSettings(ctx)
-			case watchEventTypeDelServerDir:
-				if g, ok := w.guests[guestId]; ok {
-					// this is needed for containers
-					g.ClearSettings(ctx)
-					delete(w.guests, guestId)
-				}
-				log.Infof("guest path deleted: %s", guestPath)
-			case watchEventTypeUpdServer:
-				if g, ok := w.guests[guestId]; ok {
+			} else {
+				log.Debugf("to handle inotify event %s %s", ev, wev)
+				guestId := wev.guestId
+				guestPath := wev.guestPath
+				switch wev.evType {
+				case watchEventTypeAddServerDir:
+					log.Infof("received guest path add event: %s", guestPath)
+					g, err := w.addGuestWatch(guestId, guestPath)
+					if err != nil {
+						log.Errorf("watch guest failed: %s: %s", guestPath, err)
+					}
 					g.UpdateSettings(ctx)
-				} else {
-					log.Warningf("unexpected guest update event: %s", guestPath)
-				}
-			case watchEventTypeDelServer:
-				if g, ok := w.guests[guestId]; ok {
-					log.Infof("remove guest settings %s", guestId)
-					g.ClearSettings(ctx)
-				} else {
-					log.Warningf("unexpected guest down event: %s", guestPath)
+				case watchEventTypeDelServerDir:
+					if g, ok := w.guests[guestId]; ok {
+						// this is needed for containers
+						g.ClearSettings(ctx)
+						delete(w.guests, guestId)
+					}
+					log.Infof("guest path deleted: %s", guestPath)
+				case watchEventTypeUpdServer:
+					if g, ok := w.guests[guestId]; ok {
+						g.UpdateSettings(ctx)
+					} else {
+						log.Warningf("unexpected guest update event: %s", guestPath)
+					}
+				case watchEventTypeDelServer:
+					if g, ok := w.guests[guestId]; ok {
+						log.Infof("remove guest settings %s", guestId)
+						g.ClearSettings(ctx)
+					} else {
+						log.Warningf("unexpected guest down event: %s", guestPath)
+					}
 				}
 			}
 		case <-pendingChan:
@@ -267,11 +287,16 @@ func (w *serversWatcher) Start(ctx context.Context, agent *AgentServer) {
 			log.Infof("watcher refresh time ;)")
 			w.withWait(ctx, func(ctx context.Context) {
 				w.hostLocal.UpdateSettings(ctx)
-				for _, g := range w.guests {
-					g.UpdateSettings(ctx)
-				}
+				w.scan(ctx)
+				// for _, g := range w.guests {
+				//	g.UpdateSettings(ctx)
+				// }
 			})
-		case err := <-w.watcher.Errors:
+		case err, ok := <-w.watcher.Errors:
+			if !ok {
+				log.Errorf("fsnotity.watch.Errors error")
+				goto out
+			}
 			// fail fast and recover fresh
 			panic("watcher error: %s" + err.Error())
 			return