diff --git a/charts/fission-all/templates/deployment.yaml b/charts/fission-all/templates/deployment.yaml index 51bf77a0..210ef183 100644 --- a/charts/fission-all/templates/deployment.yaml +++ b/charts/fission-all/templates/deployment.yaml @@ -72,6 +72,7 @@ rules: - services - serviceaccounts - namespaces + - events verbs: - create - delete @@ -211,8 +212,17 @@ rules: resources: - configmaps - secrets + - pods verbs: - get + - list + - watch +- apiGroups: + - "" + resources: + - events + verbs: + - '*' - apiGroups: - fission.io resources: diff --git a/cmd/builder/Dockerfile.fission-builder b/cmd/builder/Dockerfile.fission-builder index 502c3b97..8f795641 100644 --- a/cmd/builder/Dockerfile.fission-builder +++ b/cmd/builder/Dockerfile.fission-builder @@ -1,4 +1,4 @@ -FROM golang:1.12-alpine as godep +FROM golang:1.14-alpine as godep RUN apk add bash ca-certificates git gcc g++ libc-dev ARG GOPKG=github.com/fission/fission diff --git a/cmd/fetcher/Dockerfile.fission-fetcher b/cmd/fetcher/Dockerfile.fission-fetcher index ae70951a..07f66bc6 100644 --- a/cmd/fetcher/Dockerfile.fission-fetcher +++ b/cmd/fetcher/Dockerfile.fission-fetcher @@ -1,4 +1,4 @@ -FROM golang:1.12-alpine as godep +FROM golang:1.14-alpine as godep RUN apk add bash ca-certificates git gcc g++ libc-dev ARG GOPKG=github.com/fission/fission diff --git a/cmd/fetcher/app/server.go b/cmd/fetcher/app/server.go index 04c80a6e..bb1e3c68 100644 --- a/cmd/fetcher/app/server.go +++ b/cmd/fetcher/app/server.go @@ -115,6 +115,8 @@ func Run(logger *zap.Logger) { mux.HandleFunc("/specialize", f.SpecializeHandler) mux.HandleFunc("/upload", f.UploadHandler) mux.HandleFunc("/version", f.VersionHandler) + mux.HandleFunc("/wsevent/start", f.WsStartHandler) + mux.HandleFunc("/wsevent/end", f.WsEndHandler) readinessHandler := func(w http.ResponseWriter, r *http.Request) { if !*specializeOnStart || readyToServe { diff --git a/cmd/fission-bundle/Dockerfile.fission-bundle b/cmd/fission-bundle/Dockerfile.fission-bundle index 98f9f411..1526aee5 100644 --- a/cmd/fission-bundle/Dockerfile.fission-bundle +++ b/cmd/fission-bundle/Dockerfile.fission-bundle @@ -1,4 +1,4 @@ -FROM golang:1.12-alpine as godep +FROM golang:1.14-alpine as godep RUN apk add bash ca-certificates git gcc g++ libc-dev ARG GOPKG=github.com/fission/fission diff --git a/cmd/preupgradechecks/Dockerfile.fission-preupgradechecks b/cmd/preupgradechecks/Dockerfile.fission-preupgradechecks index f6c32061..2bc7463b 100644 --- a/cmd/preupgradechecks/Dockerfile.fission-preupgradechecks +++ b/cmd/preupgradechecks/Dockerfile.fission-preupgradechecks @@ -1,4 +1,4 @@ -FROM golang:1.12-alpine as godep +FROM golang:1.14-alpine as godep RUN apk add bash ca-certificates git gcc g++ libc-dev ARG GOPKG=github.com/fission/fission diff --git a/fetcher-pod-sa.yaml b/fetcher-pod-sa.yaml new file mode 100644 index 00000000..60d25848 --- /dev/null +++ b/fetcher-pod-sa.yaml @@ -0,0 +1,30 @@ +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + namespace: fission-function + name: event-fetcher +rules: +- apiGroups: [""] # "" indicates the core API group + resources: ["pods"] + verbs: ["get", "watch", "list"] +- apiGroups: [""] # "" indicates the core API group + resources: ["events"] + verbs: ["*"] + +--- + +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: fission-fetcher-pod-reader + namespace: fission-function +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: event-fetcher +subjects: +- kind: ServiceAccount + name: fission-fetcher + namespace: fission-function +--- + diff --git a/go.mod b/go.mod index a8d275e5..54793f58 100644 --- a/go.mod +++ b/go.mod @@ -62,10 +62,10 @@ require ( golang.org/x/time v0.0.0-20200630173020-3af7569d3a1e // indirect gopkg.in/jcmturner/goidentity.v3 v3.0.0 // indirect k8s.io/api v0.17.2 - k8s.io/apiextensions-apiserver v0.17.0 + k8s.io/apiextensions-apiserver v0.17.2 k8s.io/apimachinery v0.17.2 k8s.io/client-go v0.17.2 k8s.io/klog v1.0.0 - k8s.io/klog/v2 v2.5.0 + k8s.io/klog/v2 v2.8.0 k8s.io/metrics v0.17.2 ) diff --git a/go.sum b/go.sum index 6740f1bc..e2266ecf 100644 --- a/go.sum +++ b/go.sum @@ -1137,29 +1137,25 @@ honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt honnef.co/go/tools v0.0.1-2020.1.3/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k= honnef.co/go/tools v0.0.1-2020.1.4 h1:UoveltGrhghAA7ePc+e+QYDHXrBps2PqFZiHkGR/xK8= honnef.co/go/tools v0.0.1-2020.1.4/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k= -k8s.io/api v0.17.0/go.mod h1:npsyOePkeP0CPwyGfXDHxvypiYMJxBWAMpQxCaJ4ZxI= k8s.io/api v0.17.2 h1:NF1UFXcKN7/OOv1uxdRz3qfra8AHsPav5M93hlV9+Dc= k8s.io/api v0.17.2/go.mod h1:BS9fjjLc4CMuqfSO8vgbHPKMt5+SF0ET6u/RVDihTo4= -k8s.io/apiextensions-apiserver v0.17.0 h1:+XgcGxqaMztkbbvsORgCmHIb4uImHKvTjNyu7b8gRnA= -k8s.io/apiextensions-apiserver v0.17.0/go.mod h1:XiIFUakZywkUl54fVXa7QTEHcqQz9HG55nHd1DCoHj8= -k8s.io/apimachinery v0.17.0/go.mod h1:b9qmWdKlLuU9EBh+06BtLcSf/Mu89rWL33naRxs1uZg= +k8s.io/apiextensions-apiserver v0.17.2 h1:cP579D2hSZNuO/rZj9XFRzwJNYb41DbNANJb6Kolpss= +k8s.io/apiextensions-apiserver v0.17.2/go.mod h1:4KdMpjkEjjDI2pPfBA15OscyNldHWdBCfsWMDWAmSTs= k8s.io/apimachinery v0.17.2 h1:hwDQQFbdRlpnnsR64Asdi55GyCaIP/3WQpMmbNBeWr4= k8s.io/apimachinery v0.17.2/go.mod h1:b9qmWdKlLuU9EBh+06BtLcSf/Mu89rWL33naRxs1uZg= -k8s.io/apiserver v0.17.0/go.mod h1:ABM+9x/prjINN6iiffRVNCBR2Wk7uY4z+EtEGZD48cg= -k8s.io/client-go v0.17.0/go.mod h1:TYgR6EUHs6k45hb6KWjVD6jFZvJV4gHDikv/It0xz+k= +k8s.io/apiserver v0.17.2/go.mod h1:lBmw/TtQdtxvrTk0e2cgtOxHizXI+d0mmGQURIHQZlo= k8s.io/client-go v0.17.2 h1:ndIfkfXEGrNhLIgkr0+qhRguSD3u6DCmonepn1O6NYc= k8s.io/client-go v0.17.2/go.mod h1:QAzRgsa0C2xl4/eVpeVAZMvikCn8Nm81yqVx3Kk9XYI= -k8s.io/code-generator v0.17.0/go.mod h1:DVmfPQgxQENqDIzVR2ddLXMH34qeszkKSdH/N+s+38s= k8s.io/code-generator v0.17.2/go.mod h1:DVmfPQgxQENqDIzVR2ddLXMH34qeszkKSdH/N+s+38s= -k8s.io/component-base v0.17.0/go.mod h1:rKuRAokNMY2nn2A6LP/MiwpoaMRHpfRnrPaUJJj1Yoc= +k8s.io/component-base v0.17.2/go.mod h1:zMPW3g5aH7cHJpKYQ/ZsGMcgbsA/VyhEugF3QT1awLs= k8s.io/gengo v0.0.0-20190128074634-0689ccc1d7d6/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8IAqLxYwwyPxAX1Pzy0ii0= k8s.io/gengo v0.0.0-20190822140433-26a664648505/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8IAqLxYwwyPxAX1Pzy0ii0= k8s.io/klog v0.0.0-20181102134211-b9b56d5dfc92/go.mod h1:Gq+BEi5rUBO/HRz0bTSXDUcqjScdoY3a9IHpCEIOOfk= k8s.io/klog v0.3.0/go.mod h1:Gq+BEi5rUBO/HRz0bTSXDUcqjScdoY3a9IHpCEIOOfk= k8s.io/klog v1.0.0 h1:Pt+yjF5aB1xDSVbau4VsWe+dQNzA0qv1LlXdC2dF6Q8= k8s.io/klog v1.0.0/go.mod h1:4Bi6QPql/J/LkTDqv7R/cd3hPo4k2DG6Ptcz060Ez5I= -k8s.io/klog/v2 v2.5.0 h1:8mOnjf1RmUPW6KRqQCfYSZq/K20Unmp3IhuZUhxl8KI= -k8s.io/klog/v2 v2.5.0/go.mod h1:hy9LJ/NvuK+iVyP4Ehqva4HxZG/oXyIS3n3Jmire4Ec= +k8s.io/klog/v2 v2.8.0 h1:Q3gmuM9hKEjefWFFYF0Mat+YyFJvsUyYuwyNNJ5C9Ts= +k8s.io/klog/v2 v2.8.0/go.mod h1:hy9LJ/NvuK+iVyP4Ehqva4HxZG/oXyIS3n3Jmire4Ec= k8s.io/kube-openapi v0.0.0-20191107075043-30be4d16710a h1:UcxjrRMyNx/i/y8G7kPvLyy7rfbeuf1PYyBf973pgyU= k8s.io/kube-openapi v0.0.0-20191107075043-30be4d16710a/go.mod h1:1TqjTSzOxsLGIKfj0lK8EeCP7K1iUG65v09OM0/WG5E= k8s.io/metrics v0.17.2 h1:cuN1ScyUS9/tj4YFI8d0/7yO0BveFHhyQpPNWS8uLr8= diff --git a/pkg/apis/core/v1/const.go b/pkg/apis/core/v1/const.go index b5706eab..884eaae6 100644 --- a/pkg/apis/core/v1/const.go +++ b/pkg/apis/core/v1/const.go @@ -66,6 +66,8 @@ const ( SharedVolumePackages = "packages" SharedVolumeSecrets = "secrets" SharedVolumeConfigmaps = "configmaps" + PodInfoVolume = "podinfo" + PodInfoMount = "/etc/podinfo" ) const ( diff --git a/pkg/executor/executortype/poolmgr/gp.go b/pkg/executor/executortype/poolmgr/gp.go index 293b920a..4a760c72 100644 --- a/pkg/executor/executortype/poolmgr/gp.go +++ b/pkg/executor/executortype/poolmgr/gp.go @@ -77,7 +77,8 @@ type ( readyPodQueue workqueue.DelayingInterface poolInstanceID string // small random string to uniquify pod names instanceID string // poolmgr instance id - podFSVCMap sync.Map + // TODO: move this field into fsCache + podFSVCMap sync.Map } ) @@ -688,6 +689,11 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac Atime: time.Now(), } + if gp.fsCache.PodToFsvc == nil { + gp.fsCache.PodToFsvc = make(map[string]*fscache.FuncSvc) + } + gp.fsCache.PodToFsvc[pod.GetObjectMeta().GetName()] = fsvc + gp.podFSVCMap.Store(pod.ObjectMeta.Name, []interface{}{crd.CacheKey(fsvc.Function), fsvc.Address}) gp.fsCache.AddFunc(*fsvc) diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index 2b4b3ea7..cc323768 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -32,10 +32,13 @@ import ( k8serrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/runtime" k8sTypes "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/watch" k8sInformers "k8s.io/client-go/informers" "k8s.io/client-go/kubernetes" k8sCache "k8s.io/client-go/tools/cache" + k8scache "k8s.io/client-go/tools/cache" metricsclient "k8s.io/metrics/pkg/client/clientset/versioned" fv1 "github.com/fission/fission/pkg/apis/core/v1" @@ -598,6 +601,11 @@ func (gpm *GenericPoolManager) getEnvPoolsize(env *fv1.Environment) int32 { func (gpm *GenericPoolManager) idleObjectReaper() { pollSleep := 5 * time.Second + + go gpm.WebsocketStartEventChecker(gpm.kubernetesClient) + + go gpm.NoActiveConnectionEventChecker(gpm.kubernetesClient) + for { time.Sleep(pollSleep) @@ -636,6 +644,9 @@ func (gpm *GenericPoolManager) idleObjectReaper() { continue } + if _, ok := gpm.fsCache.WebsocketFsvc[fsvc.Name]; ok { + continue + } // For function with the environment that no longer exists, executor // cleanups the idle pod as usual and prints log to notify user. if _, ok := envList[fsvc.Environment.ObjectMeta.UID]; !ok { @@ -682,3 +693,87 @@ func (gpm *GenericPoolManager) idleObjectReaper() { } } } + +// WebsocketStartEventChecker checks if the pod has emitted a websocket connection start event +func (gpm *GenericPoolManager) WebsocketStartEventChecker(kubeClient *kubernetes.Clientset) { + + informer := k8scache.NewSharedInformer( + &k8scache.ListWatch{ + ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { + options.FieldSelector = "involvedObject.kind=Pod,type=Normal,reason=WsConnectionStarted" + return kubeClient.CoreV1().Events(apiv1.NamespaceAll).List(options) + }, + WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { + options.FieldSelector = "involvedObject.kind=Pod,type=Normal,reason=WsConnectionStarted" + return kubeClient.CoreV1().Events(apiv1.NamespaceAll).Watch(options) + }, + }, + &apiv1.Event{}, + 0, + ) + + stopper := make(chan struct{}) + defer close(stopper) + informer.AddEventHandler(k8scache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + mObj := obj.(metav1.Object) + gpm.logger.Info("Websocket event detected for pod", + zap.String("Pod name", mObj.GetName())) + + podName := strings.SplitAfter(mObj.GetName(), ".") + if fsvc, ok := gpm.fsCache.PodToFsvc[strings.TrimSuffix(podName[0], ".")]; ok { + gpm.fsCache.WebsocketFsvc[fsvc.Name] = true + } + }, + }) + informer.Run(stopper) + +} + +// NoActiveConnectionEventChecker checks if the pod has emitted an inactive event +func (gpm *GenericPoolManager) NoActiveConnectionEventChecker(kubeClient *kubernetes.Clientset) { + + informer := k8scache.NewSharedInformer( + &k8scache.ListWatch{ + ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { + options.FieldSelector = "involvedObject.kind=Pod,type=Normal,reason=NoActiveConnections" + return kubeClient.CoreV1().Events(apiv1.NamespaceAll).List(options) + }, + WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { + options.FieldSelector = "involvedObject.kind=Pod,type=Normal,reason=NoActiveConnections" + return kubeClient.CoreV1().Events(apiv1.NamespaceAll).Watch(options) + }, + }, + &apiv1.Event{}, + 0, + ) + + stopper := make(chan struct{}) + defer close(stopper) + informer.AddEventHandler(k8scache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + mObj := obj.(metav1.Object) + gpm.logger.Info("Inactive event detected for pod", + zap.String("Pod name", mObj.GetName())) + + podName := strings.SplitAfter(mObj.GetName(), ".") + if fsvc, ok := gpm.fsCache.PodToFsvc[strings.TrimSuffix(podName[0], ".")]; ok { + + gpm.fsCache.DeleteFunctionSvc(fsvc) + for i := range fsvc.KubernetesObjects { + gpm.logger.Info("release idle function resources due to inactivity", + zap.String("function", fsvc.Function.Name), + zap.String("address", fsvc.Address), + zap.String("executor", string(fsvc.Executor)), + zap.String("pod", fsvc.Name), + ) + reaper.CleanupKubeObject(gpm.logger, gpm.kubernetesClient, &fsvc.KubernetesObjects[i]) + time.Sleep(50 * time.Millisecond) + } + } + + }, + }) + informer.Run(stopper) + +} diff --git a/pkg/executor/fscache/functionServiceCache.go b/pkg/executor/fscache/functionServiceCache.go index b2c4e4e3..578b4740 100644 --- a/pkg/executor/fscache/functionServiceCache.go +++ b/pkg/executor/fscache/functionServiceCache.go @@ -68,8 +68,9 @@ type ( byAddress *cache.Cache // address -> function : map[string]metav1.ObjectMeta byFunctionUID *cache.Cache // function uid -> function : map[string]metav1.ObjectMeta connFunctionCache *poolcache.Cache // function-key -> funcSvc : map[string]*funcSvc - - requestChannel chan *fscRequest + PodToFsvc map[string]*FuncSvc + WebsocketFsvc map[string]bool + requestChannel chan *fscRequest } fscRequest struct { @@ -110,6 +111,8 @@ func MakeFunctionServiceCache(logger *zap.Logger) *FunctionServiceCache { byFunctionUID: cache.MakeCache(0, 0), connFunctionCache: poolcache.NewPoolCache(), requestChannel: make(chan *fscRequest), + PodToFsvc: make(map[string]*FuncSvc), + WebsocketFsvc: make(map[string]bool), } go fsc.service() return fsc diff --git a/pkg/fetcher/config/config.go b/pkg/fetcher/config/config.go index e2823256..562819f6 100644 --- a/pkg/fetcher/config/config.go +++ b/pkg/fetcher/config/config.go @@ -176,6 +176,27 @@ func (cfg *Config) fetcherCommand(extraArgs ...string) []string { } func (cfg *Config) volumesWithMounts() ([]apiv1.Volume, []apiv1.VolumeMount) { + + items := make([]apiv1.DownwardAPIVolumeFile, 0) + podNameFieldSelector := apiv1.ObjectFieldSelector{ + FieldPath: "metadata.name", + } + + podNamespaceFieldSelector := apiv1.ObjectFieldSelector{ + FieldPath: "metadata.namespace", + } + podName := apiv1.DownwardAPIVolumeFile{ + Path: "name", + FieldRef: &podNameFieldSelector, + } + + podNamespace := apiv1.DownwardAPIVolumeFile{ + Path: "namespace", + FieldRef: &podNamespaceFieldSelector, + } + + items = append(items, podName, podNamespace) + dwAPIVol := apiv1.DownwardAPIVolumeSource{Items: items} volumes := []apiv1.Volume{ { Name: fv1.SharedVolumeUserfunc, @@ -195,6 +216,12 @@ func (cfg *Config) volumesWithMounts() ([]apiv1.Volume, []apiv1.VolumeMount) { EmptyDir: &apiv1.EmptyDirVolumeSource{}, }, }, + { + Name: fv1.PodInfoVolume, + VolumeSource: apiv1.VolumeSource{ + DownwardAPI: &dwAPIVol, + }, + }, } mounts := []apiv1.VolumeMount{ { @@ -209,6 +236,10 @@ func (cfg *Config) volumesWithMounts() ([]apiv1.Volume, []apiv1.VolumeMount) { Name: fv1.SharedVolumeConfigmaps, MountPath: cfg.sharedCfgMapPath, }, + { + Name: fv1.PodInfoVolume, + MountPath: fv1.PodInfoMount, + }, } return volumes, mounts diff --git a/pkg/fetcher/fetcher.go b/pkg/fetcher/fetcher.go index 6e43cf6b..7d564971 100644 --- a/pkg/fetcher/fetcher.go +++ b/pkg/fetcher/fetcher.go @@ -32,9 +32,15 @@ import ( uuid "github.com/satori/go.uuid" "go.opencensus.io/plugin/ochttp" "go.uber.org/zap" + corev1 "k8s.io/api/core/v1" k8serr "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes" + "k8s.io/client-go/kubernetes/scheme" + typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1" + "k8s.io/client-go/tools/record" + "k8s.io/client-go/tools/reference" + "k8s.io/klog" fv1 "github.com/fission/fission/pkg/apis/core/v1" "github.com/fission/fission/pkg/crd" @@ -54,6 +60,11 @@ type ( fissionClient *crd.FissionClient kubeClient *kubernetes.Clientset httpClient *http.Client + Info PodInfo + } + PodInfo struct { + Name string + Namespace string } ) @@ -80,6 +91,17 @@ func MakeFetcher(logger *zap.Logger, sharedVolumePath string, sharedSecretPath s if err != nil { return nil, errors.Wrap(err, "error making the fission / kube client") } + + name, err := ioutil.ReadFile(fv1.PodInfoMount + "/name") + if err != nil { + return nil, errors.Wrap(err, "error reading pod name from downward volume") + } + + namespace, err := ioutil.ReadFile(fv1.PodInfoMount + "/namespace") + if err != nil { + return nil, errors.Wrap(err, "error reading pod namespace from downward volume") + } + return &Fetcher{ logger: fLogger, sharedVolumePath: sharedVolumePath, @@ -87,6 +109,10 @@ func MakeFetcher(logger *zap.Logger, sharedVolumePath string, sharedSecretPath s sharedConfigPath: sharedConfigPath, fissionClient: fissionClient, kubeClient: kubeClient, + Info: PodInfo{ + Name: string(name), + Namespace: string(namespace), + }, httpClient: &http.Client{ Transport: &ochttp.Transport{}, }, @@ -662,3 +688,77 @@ func (fetcher *Fetcher) SpecializePod(ctx context.Context, fetchReq FunctionFetc return errors.Wrapf(err, "error specializing function pod after %v times", maxRetries) } + +// WsStartHandler is used to generate websocket events in Kubernetes +func (fetcher *Fetcher) WsStartHandler(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" { + http.Error(w, "only GET is supported on this endpoint", http.StatusMethodNotAllowed) + return + } + rec, err := eventRecorder(fetcher.kubeClient) + if err != nil { + klog.Errorf("Error creating recorder %s", err) + http.Error(w, err.Error(), http.StatusInternalServerError) + } + pods, err := fetcher.kubeClient.CoreV1().Pods(fetcher.Info.Namespace).List(metav1.ListOptions{ + FieldSelector: "metadata.name=" + fetcher.Info.Name, + }) + if err != nil { + fetcher.logger.Error("Failed to get the pod", zap.Error(err)) + http.Error(w, err.Error(), http.StatusInternalServerError) + } + for _, pod := range pods.Items { + ref, err := reference.GetReference(scheme.Scheme, &pod) + if err != nil { + fetcher.logger.Error("Could not get reference for pod", zap.Error(err)) + http.Error(w, err.Error(), http.StatusInternalServerError) + } + rec.Event(ref, corev1.EventTypeNormal, "WsConnectionStarted", "Websocket connection has been formed on this pod") + fetcher.logger.Info("Sent websocket initiation event") + } + w.WriteHeader(http.StatusOK) +} + +// WsEndHandler is used to generate inactive events in Kubernetes +func (fetcher *Fetcher) WsEndHandler(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" { + http.Error(w, "only GET is supported on this endpoint", http.StatusMethodNotAllowed) + return + } + rec, err := eventRecorder(fetcher.kubeClient) + if err != nil { + klog.Errorf("Error creating recorder %s", err) + http.Error(w, err.Error(), http.StatusInternalServerError) + } + pods, err := fetcher.kubeClient.CoreV1().Pods(fetcher.Info.Namespace).List(metav1.ListOptions{ + FieldSelector: "metadata.name=" + fetcher.Info.Name, + }) + if err != nil { + fetcher.logger.Error("Failed to get the pod", zap.Error(err)) + http.Error(w, err.Error(), http.StatusInternalServerError) + } + for _, pod := range pods.Items { + // There will only be one time since we've used field selector + ref, err := reference.GetReference(scheme.Scheme, &pod) + if err != nil { + fetcher.logger.Error("Could not get reference for pod", zap.Error(err)) + http.Error(w, err.Error(), http.StatusInternalServerError) + } + // We could use Eventf and supply the amount of time the connection was inactive although, in case of multiple connections, it doesn't make sense + rec.Event(ref, corev1.EventTypeNormal, "NoActiveConnections", "Connection has been inactive") + fetcher.logger.Info("Sent no active connections event") + } + w.WriteHeader(http.StatusOK) +} + +func eventRecorder(kubeClient *kubernetes.Clientset) (record.EventRecorder, error) { + eventBroadcaster := record.NewBroadcaster() + eventBroadcaster.StartLogging(zap.S().Infof) + eventBroadcaster.StartRecordingToSink( + &typedcorev1.EventSinkImpl{ + Interface: kubeClient.CoreV1().Events("")}) + recorder := eventBroadcaster.NewRecorder( + scheme.Scheme, + corev1.EventSource{Component: "fetcher"}) + return recorder, nil +}