Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

List fission resources in specific namespace instead of all namespace (#2604)

Browse Source
This commit is contained in:
Shubham Bansal
2022-11-08 10:25:09 +05:30
committed by GitHub
parent b71a36dc1c
commit 32bd874ab6
9 changed files with 218 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/executor/executortype/container/containermgr.go
+21 -19
View File
@@ -269,28 +269,30 @@ func (caaf *Container) RefreshFuncPods(ctx context.Context, logger *zap.Logger,
// AdoptExistingResources attempts to adopt resources for functions in all namespaces.
func (caaf *Container) AdoptExistingResources(ctx context.Context) {
fnList, err := caaf.fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
if err != nil {
caaf.logger.Error("error getting function list", zap.Error(err))
return
}
wg := &sync.WaitGroup{}
for i := range fnList.Items {
fn := &fnList.Items[i]
if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypeContainer {
wg.Add(1)
go func() {
defer wg.Done()
for _, namepsace := range utils.GetNamespaces() {
fnList, err := caaf.fissionClient.CoreV1().Functions(namepsace).List(ctx, metav1.ListOptions{})
if err != nil {
caaf.logger.Error("error getting function list", zap.Error(err))
return
}
_, err = caaf.fnCreate(ctx, fn)
if err != nil {
caaf.logger.Warn("failed to adopt resources for function", zap.Error(err))
return
}
caaf.logger.Info("adopt resources for function", zap.String("function", fn.ObjectMeta.Name))
}()
for i := range fnList.Items {
fn := &fnList.Items[i]
if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypeContainer {
wg.Add(1)
go func() {
defer wg.Done()
_, err = caaf.fnCreate(ctx, fn)
if err != nil {
caaf.logger.Warn("failed to adopt resources for function", zap.Error(err))
return
}
caaf.logger.Info("adopt resources for function", zap.String("function", fn.ObjectMeta.Name))
}()
}
}
}
pkg/executor/executortype/newdeploy/newdeploymgr.go
+22 -20
View File
@@ -290,28 +290,30 @@ func (deploy *NewDeploy) RefreshFuncPods(ctx context.Context, logger *zap.Logger
// AdoptExistingResources attempts to adopt resources for functions in all namespaces.
func (deploy *NewDeploy) AdoptExistingResources(ctx context.Context) {
fnList, err := deploy.fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
if err != nil {
deploy.logger.Error("error getting function list", zap.Error(err))
return
}
wg := &sync.WaitGroup{}
for i := range fnList.Items {
fn := &fnList.Items[i]
if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypeNewdeploy {
wg.Add(1)
go func() {
defer wg.Done()
for _, namepsace := range utils.GetNamespaces() {
fnList, err := deploy.fissionClient.CoreV1().Functions(namepsace).List(ctx, metav1.ListOptions{})
if err != nil {
deploy.logger.Error("error getting function list", zap.Error(err))
return
}
_, err = deploy.fnCreate(ctx, fn)
if err != nil {
deploy.logger.Warn("failed to adopt resources for function", zap.Error(err))
return
}
deploy.logger.Info("adopt resources for function", zap.String("function", fn.ObjectMeta.Name))
}()
for i := range fnList.Items {
fn := &fnList.Items[i]
if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypeNewdeploy {
wg.Add(1)
go func() {
defer wg.Done()
_, err = deploy.fnCreate(ctx, fn)
if err != nil {
deploy.logger.Warn("failed to adopt resources for function", zap.Error(err))
return
}
deploy.logger.Info("adopt resources for function", zap.String("function", fn.ObjectMeta.Name))
}()
}
}
}
@@ -349,7 +351,7 @@ func (deploy *NewDeploy) CleanupOldExecutorObjects(ctx context.Context) {
}
func (deploy *NewDeploy) getEnvFunctions(ctx context.Context, m *metav1.ObjectMeta) []fv1.Function {
funcList, err := deploy.fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
funcList, err := deploy.fissionClient.CoreV1().Functions(m.Namespace).List(ctx, metav1.ListOptions{})
if err != nil {
deploy.logger.Error("Error getting functions for env", zap.Error(err), zap.Any("environment", m))
}
pkg/executor/executortype/poolmgr/gpm.go
+111 -107
View File
@@ -289,7 +289,7 @@ func (gpm *GenericPoolManager) RefreshFuncPods(ctx context.Context, logger *zap.
funcLabels := gp.labelsForFunction(&f.ObjectMeta)
podList, err := gpm.kubernetesClient.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{
podList, err := gpm.kubernetesClient.CoreV1().Pods(f.Spec.Environment.Namespace).List(ctx, metav1.ListOptions{
LabelSelector: labels.Set(funcLabels).AsSelector().String(),
})
@@ -311,132 +311,136 @@ func (gpm *GenericPoolManager) RefreshFuncPods(ctx context.Context, logger *zap.
}
func (gpm *GenericPoolManager) AdoptExistingResources(ctx context.Context) {
envs, err := gpm.fissionClient.CoreV1().Environments(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
if err != nil {
gpm.logger.Error("error getting environment list", zap.Error(err))
return
}
envMap := make(map[string]fv1.Environment, len(envs.Items))
envMap := make(map[string]fv1.Environment)
wg := &sync.WaitGroup{}
for i := range envs.Items {
env := envs.Items[i]
if getEnvPoolSize(&env) > 0 {
wg.Add(1)
go func() {
defer wg.Done()
_, created, err := gpm.getPool(ctx, &env)
if err != nil {
gpm.logger.Error("adopt pool failed", zap.Error(err))
}
if created {
gpm.logger.Info("created pool for the environment", zap.String("env", env.ObjectMeta.Name), zap.String("namespace", gpm.namespace))
}
}()
for _, namespace := range utils.GetNamespaces() {
envs, err := gpm.fissionClient.CoreV1().Environments(namespace).List(ctx, metav1.ListOptions{})
if err != nil {
gpm.logger.Error("error getting environment list", zap.Error(err))
return
}
// create environment map for later use
key := fmt.Sprintf("%v/%v", env.ObjectMeta.Namespace, env.ObjectMeta.Name)
envMap[key] = env
for i := range envs.Items {
env := envs.Items[i]
if getEnvPoolSize(&env) > 0 {
wg.Add(1)
go func() {
defer wg.Done()
_, created, err := gpm.getPool(ctx, &env)
if err != nil {
gpm.logger.Error("adopt pool failed", zap.Error(err))
}
if created {
gpm.logger.Info("created pool for the environment", zap.String("env", env.ObjectMeta.Name), zap.String("namespace", gpm.namespace))
}
}()
}
// create environment map for later use
key := fmt.Sprintf("%v/%v", env.ObjectMeta.Namespace, env.ObjectMeta.Name)
envMap[key] = env
}
}
l := map[string]string{
fv1.EXECUTOR_TYPE: string(fv1.ExecutorTypePoolmgr),
}
podList, err := gpm.kubernetesClient.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{
LabelSelector: labels.Set(l).AsSelector().String(),
})
for _, namespace := range utils.GetNamespaces() {
podList, err := gpm.kubernetesClient.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{
LabelSelector: labels.Set(l).AsSelector().String(),
})
if err != nil {
gpm.logger.Error("error getting pod list", zap.Error(err))
return
}
for i := range podList.Items {
pod := &podList.Items[i]
if !utils.IsReadyPod(pod) {
continue
if err != nil {
gpm.logger.Error("error getting pod list", zap.Error(err))
return
}
wg.Add(1)
go func() {
defer wg.Done()
// avoid too many requests arrive Kubernetes API server at the same time.
time.Sleep(time.Duration(rand.Intn(30)) * time.Millisecond)
patch := fmt.Sprintf(`{"metadata":{"annotations":{"%v":"%v"}}}`, fv1.EXECUTOR_INSTANCEID_LABEL, gpm.instanceID)
pod, err = gpm.kubernetesClient.CoreV1().Pods(pod.Namespace).Patch(ctx, pod.Name, k8sTypes.StrategicMergePatchType, []byte(patch), metav1.PatchOptions{})
if err != nil {
// just log the error since it won't affect the function serving
gpm.logger.Warn("error patching executor instance ID of pod", zap.Error(err),
zap.String("pod", pod.Name), zap.String("ns", pod.Namespace))
return
for i := range podList.Items {
pod := &podList.Items[i]
if !utils.IsReadyPod(pod) {
continue
}
// for unspecialized pod, we only update its annotations
if pod.Labels["managed"] == "true" {
return
}
wg.Add(1)
go func() {
defer wg.Done()
fnName, ok1 := pod.Labels[fv1.FUNCTION_NAME]
fnNS, ok2 := pod.Labels[fv1.FUNCTION_NAMESPACE]
fnUID, ok3 := pod.Labels[fv1.FUNCTION_UID]
fnRV, ok4 := pod.Annotations[fv1.FUNCTION_RESOURCE_VERSION]
envName, ok5 := pod.Labels[fv1.ENVIRONMENT_NAME]
envNS, ok6 := pod.Labels[fv1.ENVIRONMENT_NAMESPACE]
svcHost, ok7 := pod.Annotations[fv1.ANNOTATION_SVC_HOST]
env, ok8 := envMap[fmt.Sprintf("%v/%v", envNS, envName)]
// avoid too many requests arrive Kubernetes API server at the same time.
time.Sleep(time.Duration(rand.Intn(30)) * time.Millisecond)
if !(ok1 && ok2 && ok3 && ok4 && ok5 && ok6 && ok7 && ok8) {
gpm.logger.Warn("failed to adopt pod for function due to lack of necessary information",
zap.String("pod", pod.Name), zap.Any("labels", pod.Labels), zap.Any("annotations", pod.Annotations),
zap.String("env", env.ObjectMeta.Name))
return
}
fsvc := fscache.FuncSvc{
Name: pod.Name,
Function: &metav1.ObjectMeta{
Name: fnName,
Namespace: fnNS,
UID: k8sTypes.UID(fnUID),
ResourceVersion: fnRV,
},
Environment: &env,
Address: svcHost,
KubernetesObjects: []apiv1.ObjectReference{
{
Kind: "pod",
Name: pod.Name,
APIVersion: pod.TypeMeta.APIVersion,
Namespace: pod.ObjectMeta.Namespace,
ResourceVersion: pod.ObjectMeta.ResourceVersion,
UID: pod.ObjectMeta.UID,
},
},
Executor: fv1.ExecutorTypePoolmgr,
Ctime: time.Now(),
Atime: time.Now(),
}
_, err = gpm.fsCache.Add(fsvc)
if err != nil {
// If fsvc already exists we just skip the duplicate one. And let reaper to recycle the duplicate pods.
// This is for the case that there are multiple function pods for the same function due to unknown reason.
if !fscache.IsNameExistError(err) {
gpm.logger.Warn("failed to adopt pod for function", zap.Error(err), zap.String("pod", pod.Name))
patch := fmt.Sprintf(`{"metadata":{"annotations":{"%v":"%v"}}}`, fv1.EXECUTOR_INSTANCEID_LABEL, gpm.instanceID)
pod, err = gpm.kubernetesClient.CoreV1().Pods(pod.Namespace).Patch(ctx, pod.Name, k8sTypes.StrategicMergePatchType, []byte(patch), metav1.PatchOptions{})
if err != nil {
// just log the error since it won't affect the function serving
gpm.logger.Warn("error patching executor instance ID of pod", zap.Error(err),
zap.String("pod", pod.Name), zap.String("ns", pod.Namespace))
return
}
return
}
// for unspecialized pod, we only update its annotations
if pod.Labels["managed"] == "true" {
return
}
gpm.logger.Info("adopt function pod",
zap.String("pod", pod.Name), zap.Any("labels", pod.Labels), zap.Any("annotations", pod.Annotations))
}()
fnName, ok1 := pod.Labels[fv1.FUNCTION_NAME]
fnNS, ok2 := pod.Labels[fv1.FUNCTION_NAMESPACE]
fnUID, ok3 := pod.Labels[fv1.FUNCTION_UID]
fnRV, ok4 := pod.Annotations[fv1.FUNCTION_RESOURCE_VERSION]
envName, ok5 := pod.Labels[fv1.ENVIRONMENT_NAME]
envNS, ok6 := pod.Labels[fv1.ENVIRONMENT_NAMESPACE]
svcHost, ok7 := pod.Annotations[fv1.ANNOTATION_SVC_HOST]
env, ok8 := envMap[fmt.Sprintf("%v/%v", envNS, envName)]
if !(ok1 && ok2 && ok3 && ok4 && ok5 && ok6 && ok7 && ok8) {
gpm.logger.Warn("failed to adopt pod for function due to lack of necessary information",
zap.String("pod", pod.Name), zap.Any("labels", pod.Labels), zap.Any("annotations", pod.Annotations),
zap.String("env", env.ObjectMeta.Name))
return
}
fsvc := fscache.FuncSvc{
Name: pod.Name,
Function: &metav1.ObjectMeta{
Name: fnName,
Namespace: fnNS,
UID: k8sTypes.UID(fnUID),
ResourceVersion: fnRV,
},
Environment: &env,
Address: svcHost,
KubernetesObjects: []apiv1.ObjectReference{
{
Kind: "pod",
Name: pod.Name,
APIVersion: pod.TypeMeta.APIVersion,
Namespace: pod.ObjectMeta.Namespace,
ResourceVersion: pod.ObjectMeta.ResourceVersion,
UID: pod.ObjectMeta.UID,
},
},
Executor: fv1.ExecutorTypePoolmgr,
Ctime: time.Now(),
Atime: time.Now(),
}
_, err = gpm.fsCache.Add(fsvc)
if err != nil {
// If fsvc already exists we just skip the duplicate one. And let reaper to recycle the duplicate pods.
// This is for the case that there are multiple function pods for the same function due to unknown reason.
if !fscache.IsNameExistError(err) {
gpm.logger.Warn("failed to adopt pod for function", zap.Error(err), zap.String("pod", pod.Name))
}
return
}
gpm.logger.Info("adopt function pod",
zap.String("pod", pod.Name), zap.Any("labels", pod.Labels), zap.Any("annotations", pod.Annotations))
}()
}
}
wg.Wait()
pkg/executor/executortype/poolmgr/poolpodcontroller.go
+1 -1
View File
@@ -263,7 +263,7 @@ func (p *PoolPodController) workerRun(ctx context.Context, name string, processF
}
func (p *PoolPodController) getEnvLister(namespace string) (flisterv1.EnvironmentLister, error) {
lister, ok := p.envLister[metav1.NamespaceAll]
lister, ok := p.envLister[namespace]
if ok {
return lister, nil
}
pkg/executor/reaper/reaper.go
+5
View File
@@ -93,6 +93,7 @@ func CleanupDeployments(ctx context.Context, logger *zap.Logger, client kubernet
// ignore err
}
}
return nil
}
@@ -120,6 +121,7 @@ func CleanupPods(ctx context.Context, logger *zap.Logger, client kubernetes.Inte
// ignore err
}
}
return nil
}
@@ -147,6 +149,7 @@ func CleanupServices(ctx context.Context, logger *zap.Logger, client kubernetes.
// ignore err
}
}
return nil
}
@@ -175,6 +178,7 @@ func CleanupHpa(ctx context.Context, logger *zap.Logger, client kubernetes.Inter
// ignore err
}
}
return nil
}
@@ -186,6 +190,7 @@ func CleanupRoleBindings(ctx context.Context, logger *zap.Logger, client kuberne
time.Sleep(cleanupRoleBindingInterval)
logger.Debug("starting cleanupRoleBindings cycle")
// get all rolebindings ( just to be efficient, one call to kubernetes )
rbList, err := client.RbacV1().RoleBindings(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
if err != nil {