<commit_before>/* Copyright 2020 The FedLearner Authors. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package operator

import (
	"context"
	"fmt"
	"strconv"
	"strings"
	"time"

	v1 "k8s.io/api/core/v1"
	networking "k8s.io/api/networking/v1beta1"
	apiequality "k8s.io/apimachinery/pkg/api/equality"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/util/intstr"
	clientset "k8s.io/client-go/kubernetes"
	listerscorev1 "k8s.io/client-go/listers/core/v1"
	listersnetworking "k8s.io/client-go/listers/networking/v1beta1"
	"k8s.io/client-go/tools/record"
	"k8s.io/klog"

	"github.com/bytedance/fedlearner/deploy/kubernetes_operator/pkg/apis/fedlearner.k8s.io/v1alpha1"
	crdclientset "github.com/bytedance/fedlearner/deploy/kubernetes_operator/pkg/client/clientset/versioned"
	"github.com/bytedance/fedlearner/deploy/kubernetes_operator/pkg/client/clientset/versioned/scheme"
	crdlisters "github.com/bytedance/fedlearner/deploy/kubernetes_operator/pkg/client/listers/fedlearner.k8s.io/v1alpha1"
)

const (
	flReplicaTypeLabel  = "fl-replica-type"
	flReplicaIndexLabel = "fl-replica-index"
)

type AppManager interface {
	SyncApp(app *v1alpha1.FLApp, deleting bool) error
}

type appManager struct {
	namespace string

	ingressExtraHostSuffix      string
	ingressSecretName           string
	ingressEnableClientAuth     bool
	ingressClientAuthSecretName string

	kubeClient clientset.Interface
	crdClient  crdclientset.Interface

	appLister       crdlisters.FLAppLister
	configMapLister listerscorev1.ConfigMapLister
	podLister       listerscorev1.PodLister
	serviceLister   listerscorev1.ServiceLister
	ingressLister   listersnetworking.IngressLister

	appStatusUpdater StatusUpdater
	appEventHandler  AppEventHandler

	podControl     PodControlInterface
	serviceControl ServiceControlInterface

	recorder record.EventRecorder
}

var (
	_ AppManager = &appManager{}
)

func NewAppManager(
	namespace string,
	recorder record.EventRecorder,
	ingressExtraHostSuffix string,
	ingressSecretName string,
	ingressEnableClientAuth bool,
	ingressClientAuthSecretName string,
	kubeClient clientset.Interface,
	crdClient crdclientset.Interface,
	appLister crdlisters.FLAppLister,
	configMapLister listerscorev1.ConfigMapLister,
	podLister listerscorev1.PodLister,
	serviceLister listerscorev1.ServiceLister,
	ingressLister listersnetworking.IngressLister,
	appEventHandler AppEventHandler,
) AppManager {
	manager := &appManager{
		namespace: namespace,

		ingressExtraHostSuffix:      ingressExtraHostSuffix,
		ingressSecretName:           ingressSecretName,
		ingressEnableClientAuth:     ingressEnableClientAuth,
		ingressClientAuthSecretName: ingressClientAuthSecretName,

		kubeClient: kubeClient,
		crdClient:  crdClient,

		appLister:       appLister,
		configMapLister: configMapLister,
		podLister:       podLister,
		serviceLister:   serviceLister,
		ingressLister:   ingressLister,

		appStatusUpdater: NewAppStatusUpdater(crdClient, namespace),
		appEventHandler:  appEventHandler,

		podControl: RealPodControl{
			KubeClient: kubeClient,
			Recorder:   recorder,
		},
		serviceControl: RealServiceControl{
			KubeClient: kubeClient,
			Recorder:   recorder,
		},

		recorder: recorder,
	}
	return manager
}

func (am *appManager) GenOwnerReference(obj metav1.Object) *metav1.OwnerReference {
	boolPtr := func(b bool) *bool { return &b }
	controllerRef := &metav1.OwnerReference{
		APIVersion:         am.GetAPIGroupVersion().String(),
		Kind:               am.GetAPIGroupVersionKind().Kind,
		Name:               obj.GetName(),
		UID:                obj.GetUID(),
		BlockOwnerDeletion: boolPtr(true),
		Controller:         boolPtr(true),
	}

	return controllerRef
}

func (am *appManager) GetAPIGroupVersionKind() schema.GroupVersionKind {
	return v1alpha1.SchemeGroupVersionKind
}

func (am *appManager) GetAPIGroupVersion() schema.GroupVersion {
	return v1alpha1.SchemeGroupVersion
}

func (am *appManager) SyncApp(app *v1alpha1.FLApp, deleting bool) error {
	ctx := context.Background()

	appCopy := app.DeepCopy()
	name := appCopy.Name

	scheme.Scheme.Default(appCopy)
	if deleting {
		klog.Infof("deleting application %v, shut it down", name)
		return am.deleteApp(ctx, appCopy)
	}

	appState := appCopy.Status.AppState
	if appState == v1alpha1.FLStateNew || appState == v1alpha1.FLStateBootstrapped || appState == v1alpha1.FLStateSyncSent || appState == v1alpha1.FLStateRunning {
		now := time.Now()
		timeout := am.isAppTimeOut(appCopy, now)
		if am.isAppFailing(appCopy) || timeout {
			if timeout {
				klog.Errorf("app is failing because timed out, name = %v, creationTimestamp = %v, activeDeadlineSeconds = %v, now is %v", name, appCopy.GetCreationTimestamp(), *appCopy.Spec.ActiveDeadlineSeconds, now)
			}
			return am.appStatusUpdater.UpdateAppStateWithRetry(ctx, appCopy, v1alpha1.FLStateFailing)
		}
	}

	switch appState {
	case v1alpha1.FLStateNew:
		return am.syncNewApp(ctx, appCopy)
	case v1alpha1.FLStateBootstrapped:
		return am.syncBootstrappedApp(ctx, appCopy)
	case v1alpha1.FLStateSyncSent:
		return am.syncSyncSentApp(ctx, appCopy)
	case v1alpha1.FLStateRunning:
		return am.syncRunningApp(ctx, appCopy)
	case v1alpha1.FLStateFailing:
		return am.syncFailingApp(ctx, appCopy)
	case v1alpha1.FLStateShutDown:
		return am.syncShuttingDownApp(ctx, appCopy)
	case v1alpha1.FLStateComplete, v1alpha1.FLStateFailed:
		klog.Infof("ignore app %v as its state is %v", name, appState)
	}
	return nil
}

func (am *appManager) deleteApp(ctx context.Context, app *v1alpha1.FLApp) error {
	name := app.Name
	appState := app.Status.AppState
	if appState == "" {
		appState = v1alpha1.FLStateNew
	}
	switch appState {
	case v1alpha1.FLStateNew, v1alpha1.FLStateBootstrapped, v1alpha1.FLStateSyncSent, v1alpha1.FLStateRunning:
		klog.Infof("shutting down peer as app is deleted, name = %v", name)
		if err := am.appEventHandler.Shutdown(ctx, app); err != nil {<commit_msg>fix: fix completed app state (#341)<commit_after>/* Copyright 2020 The FedLearner Authors. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package operator

import (
	"context"
	"fmt"
	"strconv"
	"strings"
	"time"

	v1 "k8s.io/api/core/v1"
	networking "k8s.io/api/networking/v1beta1"
	apiequality "k8s.io/apimachinery/pkg/api/equality"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/util/intstr"
	clientset "k8s.io/client-go/kubernetes"
	listerscorev1 "k8s.io/client-go/listers/core/v1"
	listersnetworking "k8s.io/client-go/listers/networking/v1beta1"
	"k8s.io/client-go/tools/record"
	"k8s.io/klog"

	"github.com/bytedance/fedlearner/deploy/kubernetes_operator/pkg/apis/fedlearner.k8s.io/v1alpha1"
	crdclientset "github.com/bytedance/fedlearner/deploy/kubernetes_operator/pkg/client/clientset/versioned"
	"github.com/bytedance/fedlearner/deploy/kubernetes_operator/pkg/client/clientset/versioned/scheme"
	crdlisters "github.com/bytedance/fedlearner/deploy/kubernetes_operator/pkg/client/listers/fedlearner.k8s.io/v1alpha1"
)

const (
	flReplicaTypeLabel  = "fl-replica-type"
	flReplicaIndexLabel = "fl-replica-index"
)

type AppManager interface {
	SyncApp(app *v1alpha1.FLApp, deleting bool) error
}

type appManager struct {
	namespace string

	ingressExtraHostSuffix      string
	ingressSecretName           string
	ingressEnableClientAuth     bool
	ingressClientAuthSecretName string

	kubeClient clientset.Interface
	crdClient  crdclientset.Interface

	appLister       crdlisters.FLAppLister
	configMapLister listerscorev1.ConfigMapLister
	podLister       listerscorev1.PodLister
	serviceLister   listerscorev1.ServiceLister
	ingressLister   listersnetworking.IngressLister

	appStatusUpdater StatusUpdater
	appEventHandler  AppEventHandler

	podControl     PodControlInterface
	serviceControl ServiceControlInterface

	recorder record.EventRecorder
}

var (
	_ AppManager = &appManager{}
)

func NewAppManager(
	namespace string,
	recorder record.EventRecorder,
	ingressExtraHostSuffix string,
	ingressSecretName string,
	ingressEnableClientAuth bool,
	ingressClientAuthSecretName string,
	kubeClient clientset.Interface,
	crdClient crdclientset.Interface,
	appLister crdlisters.FLAppLister,
	configMapLister listerscorev1.ConfigMapLister,
	podLister listerscorev1.PodLister,
	serviceLister listerscorev1.ServiceLister,
	ingressLister listersnetworking.IngressLister,
	appEventHandler AppEventHandler,
) AppManager {
	manager := &appManager{
		namespace: namespace,

		ingressExtraHostSuffix:      ingressExtraHostSuffix,
		ingressSecretName:           ingressSecretName,
		ingressEnableClientAuth:     ingressEnableClientAuth,
		ingressClientAuthSecretName: ingressClientAuthSecretName,

		kubeClient: kubeClient,
		crdClient:  crdClient,

		appLister:       appLister,
		configMapLister: configMapLister,
		podLister:       podLister,
		serviceLister:   serviceLister,
		ingressLister:   ingressLister,

		appStatusUpdater: NewAppStatusUpdater(crdClient, namespace),
		appEventHandler:  appEventHandler,

		podControl: RealPodControl{
			KubeClient: kubeClient,
			Recorder:   recorder,
		},
		serviceControl: RealServiceControl{
			KubeClient: kubeClient,
			Recorder:   recorder,
		},

		recorder: recorder,
	}
	return manager
}

func (am *appManager) GenOwnerReference(obj metav1.Object) *metav1.OwnerReference {
	boolPtr := func(b bool) *bool { return &b }
	controllerRef := &metav1.OwnerReference{
		APIVersion:         am.GetAPIGroupVersion().String(),
		Kind:               am.GetAPIGroupVersionKind().Kind,
		Name:               obj.GetName(),
		UID:                obj.GetUID(),
		BlockOwnerDeletion: boolPtr(true),
		Controller:         boolPtr(true),
	}

	return controllerRef
}

func (am *appManager) GetAPIGroupVersionKind() schema.GroupVersionKind {
	return v1alpha1.SchemeGroupVersionKind
}

func (am *appManager) GetAPIGroupVersion() schema.GroupVersion {
	return v1alpha1.SchemeGroupVersion
}

func (am *appManager) SyncApp(app *v1alpha1.FLApp, deleting bool) error {
	ctx := context.Background()

	appCopy := app.DeepCopy()
	name := appCopy.Name

	scheme.Scheme.Default(appCopy)
	if deleting {
		klog.Infof("deleting application %v, shut it down", name)
		return am.deleteApp(ctx, appCopy)
	}

	appState := appCopy.Status.AppState
	if appState == v1alpha1.FLStateNew || appState == v1alpha1.FLStateBootstrapped || appState == v1alpha1.FLStateSyncSent || appState == v1alpha1.FLStateRunning {
		now := time.Now()
		timeout := am.isAppTimeOut(appCopy, now)
		if am.isAppFailing(appCopy) || timeout {
			if timeout {
				klog.Errorf("app is failing because timed out, name = %v, creationTimestamp = %v, activeDeadlineSeconds = %v, now is %v", name, appCopy.GetCreationTimestamp(), *appCopy.Spec.ActiveDeadlineSeconds, now)
			}
			return am.appStatusUpdater.UpdateAppStateWithRetry(ctx, appCopy, v1alpha1.FLStateFailing)
		}
	}
	if appState == v1alpha1.FLStateNew && am.isAppFinished(appCopy) {
		return am.appStatusUpdater.UpdateAppStateWithRetry(ctx, appCopy, v1alpha1.FLStateComplete)
	}

	switch appState {
	case v1alpha1.FLStateNew:
		return am.syncNewApp(ctx, appCopy)
	case v1alpha1.FLStateBootstrapped:
		return am.syncBootstrappedApp(ctx, appCopy)
	case v1alpha1.FLStateSyncSent:
		return am.syncSyncSentApp(ctx, appCopy)
	case v1alpha1.FLStateRunning:
		return am.syncRunningApp(ctx, appCopy)
	case v1alpha1.FLStateFailing:
		return am.syncFailingApp(ctx, appCopy)
	case v1alpha1.FLStateShutDown:
		return am.syncShuttingDownApp(ctx, appCopy)
	case v1alpha1.FLStateComplete, v1alpha1.FLStateFailed:
		klog.Infof("ignore app %v as its state is %v", name, appState)
	}
	return nil
}

func (am *appManager) deleteApp(ctx context.Context, app *v1alpha1.FLApp) error {
	name := app.Name
	appState := app.Status.AppState
	if appState == "" {
		appState = v1alpha1.FLStateNew
	}
	switch appState {
	case v1alpha1.FLStateNew, v1alpha1.FLStateBootstrapped, v1alpha1.FLStateSyncSent, v1alpha1.FLStateRunning:
		klog.Infof("shutting down peer as app is deleted, name = %v", name)
		if err := am.appEventHandler.Shutdown(ctx, app); err != nil {
{'fraction_non_alphanumeric': 0.092297225791821, 'fraction_numerical': 0.014286794164335929, 'mean_word_length': 4.820501539815222, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4225682', 'n_tokens_mistral': 4720, 'n_tokens_neox': 4546, 'n_words': 913}