Merge pull request #475 from lec-bit/kmesh-build

kmesh support restart by reload old bpf map and prog

build/docker/start_kmesh.sh

@@ -40,9 +40,7 @@ function cleanup(){
                   rmmod kmesh
           fi
 
-          umount -t cgroup2 /mnt/kmesh_cgroup2/
-          rm -rf /mnt/kmesh_cgroup2
-          rm -rf /sys/fs/bpf/bpf_kmesh
+        echo "kmesh close"
 }
 
 trap 'stop_kmesh' SIGTERM

daemon/manager/manager.go

@@ -105,6 +105,7 @@ func Execute(configs *options.BootstrapConfigs) error {
 	defer cniInstaller.Stop()
 
 	setupCloseHandler()
+	bpf.SetCloseStatus()
 	return nil
 }
 
@@ -113,7 +114,6 @@ func setupCloseHandler() {
 	signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT, syscall.SIGHUP, syscall.SIGABRT, syscall.SIGTSTP)
 
 	<-ch
-
 	log.Warn("exiting...")
 }
 

deploy/helm/templates/kmesh-rbac.yaml

@@ -19,6 +19,12 @@ rules:
   - patch
   - list
   - watch
+- apiGroups:
+  - "apps"
+  resources:
+  - daemonsets
+  verbs:
+  - get
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
@@ -34,4 +40,4 @@ roleRef:
 subjects:
 - kind: ServiceAccount
   name: '{{ include "kmesh.fullname" . }}'
-  namespace: '{{ .Release.Namespace }}'
+  namespace: '{{ .Release.Namespace }}'

deploy/yaml/clusterrole.yaml

@@ -8,3 +8,6 @@ rules:
 - apiGroups: [""]
   resources: ["pods","services","namespaces"]
   verbs: ["get", "update", "patch", "list", "watch"]
+- apiGroups: ["apps"]
+  resources: ["daemonsets"]
+  verbs: ["get"]

pkg/bpf/bpf.go

@@ -22,17 +22,25 @@ package bpf
 import "C"
 import (
 	"fmt"
+	"hash/fnv"
+	"os"
 	"os/exec"
+	"path/filepath"
 	"strings"
+	"syscall"
 
 	"github.com/cilium/ebpf"
 	"github.com/cilium/ebpf/rlimit"
 
 	"kmesh.net/kmesh/daemon/options"
 	"kmesh.net/kmesh/pkg/logger"
+	"kmesh.net/kmesh/pkg/version"
 )
 
-var log = logger.NewLoggerField("pkg/bpf")
+var (
+	log  = logger.NewLoggerField("pkg/bpf")
+	hash = fnv.New32a()
+)
 
 type BpfInfo struct {
 	MapPath          string
@@ -50,11 +58,13 @@ type BpfLoader struct {
 	obj         *BpfKmesh
 	workloadObj *BpfKmeshWorkload
 	bpfLogLevel *ebpf.Map
+	VersionMap  *ebpf.Map
 }
 
 func NewBpfLoader(config *options.BpfConfig) *BpfLoader {
 	return &BpfLoader{
-		config: config,
+		config:     config,
+		VersionMap: NewVersionMap(config),
 	}
 }
 
@@ -109,6 +119,10 @@ func (l *BpfLoader) Start(config *options.BpfConfig) error {
 		return err
 	}
 
+	if l.VersionMap == nil {
+		return fmt.Errorf("NewVersionMap failed")
+	}
+
 	if config.AdsEnabled() {
 		if err = l.StartAdsMode(); err != nil {
 			return err
@@ -125,6 +139,9 @@ func (l *BpfLoader) Start(config *options.BpfConfig) error {
 		}
 	}
 
+	if GetStartType() == Restart {
+		log.Infof("bpf load from last pinPath")
+	}
 	return nil
 }
 
@@ -156,6 +173,12 @@ func StopMda() error {
 
 func (l *BpfLoader) Stop() {
 	var err error
+	if GetStartType() == Restart {
+		log.Infof("kmesh restart, not clean bpf map and prog")
+		return
+	}
+
+	Close(l.VersionMap)
 
 	if l.config.AdsEnabled() {
 		C.deserial_uninit()
@@ -176,4 +199,107 @@ func (l *BpfLoader) Stop() {
 			return
 		}
 	}
+	CleanupBpfMap()
+}
+
+func NewVersionMap(config *options.BpfConfig) *ebpf.Map {
+	var versionPath string
+	if config.AdsEnabled() {
+		versionPath = filepath.Join(config.BpfFsPath + "/bpf_kmesh/map/")
+	} else if config.WdsEnabled() {
+		versionPath = filepath.Join(config.BpfFsPath + "/bpf_kmesh_workload/map/")
+	}
+
+	_, err := os.Stat(versionPath)
+	if err == nil {
+		m := recoverVersionMap(config, versionPath)
+		if m != nil {
+			SetStartStatus(m)
+			return m
+		}
+	}
+
+	mapSpec := &ebpf.MapSpec{
+		Name:       "kmesh_version",
+		Type:       ebpf.Array,
+		KeySize:    4,
+		ValueSize:  4,
+		MaxEntries: 1,
+	}
+	m, err := ebpf.NewMap(mapSpec)
+	if err != nil {
+		log.Errorf("Create kmesh_version map failed, err is %v", err)
+		return nil
+	}
+
+	if err := os.MkdirAll(versionPath,
+		syscall.S_IRUSR|syscall.S_IWUSR|syscall.S_IXUSR|syscall.S_IRGRP|syscall.S_IXGRP); err != nil && !os.IsExist(err) {
+		log.Errorf("mkdir failed %v", err)
+		return nil
+	}
+
+	err = m.Pin(filepath.Join(versionPath + "/kmesh_version"))
+	if err != nil {
+		log.Errorf("kmesh_version pin failed: %v", err)
+		return nil
+	}
+
+	storeVersionInfo(m)
+	log.Infof("kmesh start with Normal")
+	SetStartType(Normal)
+	return m
+}
+
+func storeVersionInfo(versionMap *ebpf.Map) {
+	key := uint32(0)
+	var value uint32
+	hash.Reset()
+	hash.Write([]byte(version.Get().GitVersion))
+	value = hash.Sum32()
+	if err := versionMap.Put(&key, &value); err != nil {
+		log.Errorf("Add Version Map failed, err is %v", err)
+	}
+}
+
+func getOldVersionFromMap(m *ebpf.Map, key uint32) uint32 {
+	var value uint32
+	err := m.Lookup(&key, &value)
+	if err != nil {
+		log.Errorf("lookup failed: %v", err)
+		return value
+	}
+	return value
+}
+
+func recoverVersionMap(config *options.BpfConfig, versionPath string) *ebpf.Map {
+	opts := &ebpf.LoadPinOptions{
+		ReadOnly:  false,
+		WriteOnly: false,
+		Flags:     0,
+	}
+
+	versionMap, err := ebpf.LoadPinnedMap(filepath.Join(versionPath+"/kmesh_version"), opts)
+	if err != nil {
+		log.Infof("kmesh version map loadfailed: %v, start normally", err)
+
+		return nil
+	}
+	log.Debugf("recoverVersionMap success")
+
+	return versionMap
+}
+
+func Close(m *ebpf.Map) {
+	var err error
+	err = m.Unpin()
+	if err != nil {
+		log.Errorf("Failed to Unpin kmesh_version :%v", err)
+	}
+
+	err = m.Close()
+	if err != nil {
+		log.Errorf("Failed to Close kmesh_version :%v", err)
+	}
+
+	log.Infof("Clean kmesh_version map and bpf prog")
 }

pkg/bpf/bpf_kmesh_workload.go

@@ -83,6 +83,10 @@ func (sc *BpfSockConnWorkload) loadKmeshSockConnObjects() (*ebpf.CollectionSpec,
 		return nil, err
 	}
 
+	if GetStartType() == Restart {
+		return spec, nil
+	}
+
 	value := reflect.ValueOf(sc.KmeshCgroupSockWorkloadObjects.KmeshCgroupSockWorkloadPrograms)
 	if err = pinPrograms(&value, sc.Info.BpfFsPath); err != nil {
 		return nil, err
@@ -149,6 +153,18 @@ func (sc *BpfSockConnWorkload) Attach() error {
 	}
 
 	sc.Link6, err = link.AttachCgroup(cgopt)
+
+	if GetStartType() == Restart {
+		return err
+	}
+
+	if err := sc.Link.Pin(sc.Info.BpfFsPath + "sockconn_prog"); err != nil {
+		return err
+	}
+	if err := sc.Link6.Pin(sc.Info.BpfFsPath + "sockconn6_prog"); err != nil {
+		return err
+	}
+
 	return err
 }
 
@@ -236,13 +252,18 @@ func (so *BpfSockOpsWorkload) loadKmeshSockopsObjects() (*ebpf.CollectionSpec, e
 		return nil, err
 	}
 
+	if GetStartType() == Restart {
+		return spec, nil
+	}
+
 	value := reflect.ValueOf(so.KmeshSockopsWorkloadObjects.KmeshSockopsWorkloadPrograms)
 	if err = pinPrograms(&value, so.Info.BpfFsPath); err != nil {
 		return nil, err
 	}
 
 	return spec, nil
 }
+
 func (so *BpfSockOpsWorkload) LoadSockOps() error {
 	/* load kmesh sockops main bpf prog*/
 	spec, err := so.loadKmeshSockopsObjects()
@@ -269,6 +290,14 @@ func (so *BpfSockOpsWorkload) Attach() error {
 		return err
 	}
 	so.Link = lk
+
+	if GetStartType() == Restart {
+		return nil
+	}
+
+	if err := lk.Pin(so.Info.BpfFsPath + "cgroup_sockops_prog"); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -355,6 +384,11 @@ func (sm *BpfSendMsgWorkload) loadKmeshSendmsgObjects() (*ebpf.CollectionSpec, e
 	if err = spec.LoadAndAssign(&sm.KmeshSendmsgObjects, &opts); err != nil {
 		return nil, err
 	}
+
+	if GetStartType() == Restart {
+		return spec, nil
+	}
+
 	value := reflect.ValueOf(sm.KmeshSendmsgObjects.KmeshSendmsgPrograms)
 	if err = pinPrograms(&value, sm.Info.BpfFsPath); err != nil {
 		return nil, err
@@ -474,6 +508,10 @@ func (xa *BpfXdpAuthWorkload) loadKmeshXdpAuthObjects() (*ebpf.CollectionSpec, e
 		return nil, err
 	}
 
+	if GetStartType() == Restart {
+		return spec, nil
+	}
+
 	value := reflect.ValueOf(xa.KmeshXDPAuthObjects.KmeshXDPAuthPrograms)
 	if err = pinPrograms(&value, xa.Info.BpfFsPath); err != nil {
 		return nil, err