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

Deployments to work with specific namespaces instead of all namespaces (#2635)

Browse Source 
* remove namespaceAll in deployments
* remove GetNamespace function from informer
* added logger and improvement for naming
* move reaper logic to utils

Signed-off-by: Sanket Sudake <sanketsudake@gmail.com>
Co-authored-by: Sanket Sudake <sanketsudake@gmail.com>
This commit is contained in:
Shubham Bansal
2022-11-29 17:09:51 +05:30
committed by GitHub
co-authored by Sanket Sudake
parent 0aec9e139e
commit 526b5f0beb
11 changed files with 366 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/preupgradechecks/checks.go
+1 -1
View File
@@ -123,7 +123,7 @@ func (client *PreUpgradeTaskClient) VerifyFunctionSpecReferences(ctx context.Con
var fList *fv1.FunctionList
errs := &multierror.Error{}
for _, namespace := range utils.GetNamespaces() {
for _, namespace := range utils.DefaultNSResolver().FissionResourceNS {
for i := 0; i < maxRetries; i++ {
fList, err = client.fissionClient.CoreV1().Functions(namespace).List(ctx, metav1.ListOptions{})
if err == nil {
pkg/executor/executor.go
+2 -2
View File
@@ -290,7 +290,7 @@ func StartExecutor(ctx context.Context, logger *zap.Logger, port int) error {
envInformer := make(map[string]finformerv1.EnvironmentInformer, 0)
pkgInformer := make(map[string]finformerv1.PackageInformer, 0)
for _, ns := range utils.GetNamespaces() {
for _, ns := range utils.DefaultNSResolver().FissionResourceNS {
factory := genInformer.NewFilteredSharedInformerFactory(fissionClient, time.Minute*30, ns, nil)
funcInformer[ns] = factory.Core().V1().Functions()
envInformer[ns] = factory.Core().V1().Environments()
@@ -369,7 +369,7 @@ func StartExecutor(ctx context.Context, logger *zap.Logger, port int) error {
configMapInformer := make(map[string]k8sInformersv1.ConfigMapInformer, 0)
secretInformer := make(map[string]k8sInformersv1.SecretInformer, 0)
for _, ns := range utils.GetNamespaces() {
for _, ns := range utils.DefaultNSResolver().FissionResourceNS {
factory := k8sInformers.NewFilteredSharedInformerFactory(kubernetesClient, time.Minute*30, ns, nil)
configMapInformer[ns] = factory.Core().V1().ConfigMaps()
secretInformer[ns] = factory.Core().V1().Secrets()
pkg/executor/executortype/container/containermgr.go
+3 -2
View File
@@ -242,7 +242,8 @@ func (caaf *Container) RefreshFuncPods(ctx context.Context, logger *zap.Logger,
funcLabels := caaf.getDeployLabels(f.ObjectMeta)
dep, err := caaf.kubernetesClient.AppsV1().Deployments(metav1.NamespaceAll).List(ctx, metav1.ListOptions{
nsResolver := utils.DefaultNSResolver()
dep, err := caaf.kubernetesClient.AppsV1().Deployments(nsResolver.GetFunctionNS(f.ObjectMeta.Namespace)).List(ctx, metav1.ListOptions{
LabelSelector: labels.Set(funcLabels).AsSelector().String(),
})
if err != nil {
@@ -273,7 +274,7 @@ func (caaf *Container) RefreshFuncPods(ctx context.Context, logger *zap.Logger,
func (caaf *Container) AdoptExistingResources(ctx context.Context) {
wg := &sync.WaitGroup{}
for _, namepsace := range utils.GetNamespaces() {
for _, namepsace := range utils.DefaultNSResolver().FissionResourceNS {
fnList, err := caaf.fissionClient.CoreV1().Functions(namepsace).List(ctx, metav1.ListOptions{})
if err != nil {
caaf.logger.Error("error getting function list", zap.Error(err))
pkg/executor/executortype/newdeploy/newdeploymgr.go
+3 -3
View File
@@ -260,7 +260,7 @@ func (deploy *NewDeploy) RefreshFuncPods(ctx context.Context, logger *zap.Logger
UID: env.ObjectMeta.UID,
})
dep, err := deploy.kubernetesClient.AppsV1().Deployments(metav1.NamespaceAll).List(ctx, metav1.ListOptions{
dep, err := deploy.kubernetesClient.AppsV1().Deployments(deploy.nsResolver.GetFunctionNS(f.ObjectMeta.Namespace)).List(ctx, metav1.ListOptions{
LabelSelector: labels.Set(funcLabels).AsSelector().String(),
})
@@ -292,7 +292,7 @@ func (deploy *NewDeploy) RefreshFuncPods(ctx context.Context, logger *zap.Logger
func (deploy *NewDeploy) AdoptExistingResources(ctx context.Context) {
wg := &sync.WaitGroup{}
for _, namepsace := range utils.GetNamespaces() {
for _, namepsace := range utils.DefaultNSResolver().FissionResourceNS {
fnList, err := deploy.fissionClient.CoreV1().Functions(namepsace).List(ctx, metav1.ListOptions{})
if err != nil {
deploy.logger.Error("error getting function list", zap.Error(err))
@@ -769,7 +769,7 @@ func (deploy *NewDeploy) idleObjectReaper(ctx context.Context) {
func (deploy *NewDeploy) doIdleObjectReaper(ctx context.Context) {
envList := make(map[k8sTypes.UID]struct{})
for _, namespace := range utils.GetNamespaces() {
for _, namespace := range utils.DefaultNSResolver().FissionResourceNS {
envs, err := deploy.fissionClient.CoreV1().Environments(namespace).List(ctx, metav1.ListOptions{})
if err != nil {
deploy.logger.Fatal("failed to get environment list", zap.Error(err), zap.String("namespace", namespace))
pkg/executor/executortype/newdeploy/newdeploymgr_test.go
+1
View File
@@ -92,6 +92,7 @@ func TestRefreshFuncPods(t *testing.T) {
nsResolver := utils.NamespaceResolver{
FunctionNamespace: functionNamespace,
BuiderNamespace: builderNamespace,
DefaultNamespace: defaultNamespace,
}
ndm.nsResolver = &nsResolver
pkg/executor/executortype/poolmgr/gpm.go
+4 -4
View File
@@ -317,7 +317,7 @@ func (gpm *GenericPoolManager) AdoptExistingResources(ctx context.Context) {
envMap := make(map[string]fv1.Environment)
wg := &sync.WaitGroup{}
for _, namespace := range utils.GetNamespaces() {
for _, namespace := range utils.DefaultNSResolver().FissionResourceNS {
envs, err := gpm.fissionClient.CoreV1().Environments(namespace).List(ctx, metav1.ListOptions{})
if err != nil {
gpm.logger.Error("error getting environment list", zap.Error(err))
@@ -351,7 +351,7 @@ func (gpm *GenericPoolManager) AdoptExistingResources(ctx context.Context) {
fv1.EXECUTOR_TYPE: string(fv1.ExecutorTypePoolmgr),
}
for _, namespace := range utils.GetNamespaces() {
for _, namespace := range utils.DefaultNSResolver().FissionResourceNS {
podList, err := gpm.kubernetesClient.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{
LabelSelector: labels.Set(l).AsSelector().String(),
})
@@ -588,7 +588,7 @@ func (gpm *GenericPoolManager) idleObjectReaper(ctx context.Context) {
func (gpm *GenericPoolManager) doIdleObjectReaper(ctx context.Context) {
envList := make(map[k8sTypes.UID]struct{})
for _, namespace := range utils.GetNamespaces() {
for _, namespace := range utils.DefaultNSResolver().FissionResourceNS {
envs, err := gpm.fissionClient.CoreV1().Environments(namespace).List(ctx, metav1.ListOptions{})
if err != nil {
gpm.logger.Error("failed to get environment list", zap.Error(err), zap.String("namespace", namespace))
@@ -601,7 +601,7 @@ func (gpm *GenericPoolManager) doIdleObjectReaper(ctx context.Context) {
}
fnList := make(map[k8sTypes.UID]fv1.Function)
for _, namespace := range utils.GetNamespaces() {
for _, namespace := range utils.DefaultNSResolver().FissionResourceNS {
fns, err := gpm.fissionClient.CoreV1().Functions(namespace).List(ctx, metav1.ListOptions{})
if err != nil {
gpm.logger.Error("failed to get environment list", zap.Error(err), zap.String("namespace", namespace))
pkg/executor/reaper/reaper.go
+55 -6
View File
@@ -71,7 +71,8 @@ func CleanupKubeObject(ctx context.Context, logger *zap.Logger, kubeClient kuber
// CleanupDeployments deletes deployment(s) for a given instanceID
func CleanupDeployments(ctx context.Context, logger *zap.Logger, client kubernetes.Interface, instanceID string, listOps metav1.ListOptions) error {
deploymentList, err := client.AppsV1().Deployments(metav1.NamespaceAll).List(ctx, listOps)
cleanupDeployments := func(namespace string) error {
deploymentList, err := client.AppsV1().Deployments(namespace).List(ctx, listOps)
if err != nil {
return err
}
@@ -93,13 +94,21 @@ func CleanupDeployments(ctx context.Context, logger *zap.Logger, client kubernet
// ignore err
}
}
return nil
}
for _, namespace := range GetReaperNamespace() {
if err := cleanupDeployments(namespace); err != nil {
return err
}
}
return nil
}
// CleanupPods deletes pod(s) for a given instanceID
func CleanupPods(ctx context.Context, logger *zap.Logger, client kubernetes.Interface, instanceID string, listOps metav1.ListOptions) error {
podList, err := client.CoreV1().Pods(metav1.NamespaceAll).List(ctx, listOps)
cleanupPods := func(namespace string) error {
podList, err := client.CoreV1().Pods(namespace).List(ctx, listOps)
if err != nil {
return err
}
@@ -121,13 +130,22 @@ func CleanupPods(ctx context.Context, logger *zap.Logger, client kubernetes.Inte
// ignore err
}
}
return nil
}
for _, namespace := range GetReaperNamespace() {
if err := cleanupPods(namespace); err != nil {
return err
}
}
return nil
}
// CleanupServices deletes service(s) for a given instanceID
func CleanupServices(ctx context.Context, logger *zap.Logger, client kubernetes.Interface, instanceID string, listOps metav1.ListOptions) error {
svcList, err := client.CoreV1().Services(metav1.NamespaceAll).List(ctx, listOps)
cleanupServices := func(namespace string) error {
svcList, err := client.CoreV1().Services(namespace).List(ctx, listOps)
if err != nil {
return err
}
@@ -149,13 +167,22 @@ func CleanupServices(ctx context.Context, logger *zap.Logger, client kubernetes.
// ignore err
}
}
return nil
}
for _, namespace := range GetReaperNamespace() {
if err := cleanupServices(namespace); err != nil {
return err
}
}
return nil
}
// CleanupHpa deletes horizontal pod autoscaler(s) for a given instanceID
func CleanupHpa(ctx context.Context, logger *zap.Logger, client kubernetes.Interface, instanceID string, listOps metav1.ListOptions) error {
hpaList, err := client.AutoscalingV2beta2().HorizontalPodAutoscalers(metav1.NamespaceAll).List(ctx, listOps)
cleanupHpa := func(namespace string) error {
hpaList, err := client.AutoscalingV2beta2().HorizontalPodAutoscalers(namespace).List(ctx, listOps)
if err != nil {
return err
}
@@ -178,6 +205,14 @@ func CleanupHpa(ctx context.Context, logger *zap.Logger, client kubernetes.Inter
// ignore err
}
}
return nil
}
for _, namespace := range GetReaperNamespace() {
if err := cleanupHpa(namespace); err != nil {
return err
}
}
return nil
}
@@ -192,12 +227,13 @@ func CleanupRoleBindings(ctx context.Context, logger *zap.Logger, client kuberne
logger.Debug("starting cleanupRoleBindings cycle")
cleanupRoleBindings := func(namespace string) error {
// get all rolebindings ( just to be efficient, one call to kubernetes )
rbList, err := client.RbacV1().RoleBindings(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
rbList, err := client.RbacV1().RoleBindings(namespace).List(ctx, metav1.ListOptions{})
if err != nil {
// something wrong, but next iteration hopefully succeeds
logger.Error("error listing role bindings in all namespaces", zap.Error(err))
continue
return err
}
// go through each role-binding object and do the cleanup necessary
@@ -324,5 +360,18 @@ func CleanupRoleBindings(ctx context.Context, logger *zap.Logger, client kuberne
}
}
}
return nil
}
for _, namespace := range GetReaperNamespace() {
//ignore error
cleanupRoleBindings(namespace) //nolint errcheck
}
}
}
func GetReaperNamespace() map[string]string {
ns := utils.DefaultNSResolver()
//to support backward compatibility we need to cleanup deployment and rolebinding created in function, buidler and default namespace as well
fissionResourceNs := ns.FissionNSWithOptions(utils.WithBuilderNs(), utils.WithFunctionNs(), utils.WithDefaultNs())
return fissionResourceNs
}
pkg/storagesvc/archivePruner.go
+1 -1
View File
@@ -83,7 +83,7 @@ func (pruner *ArchivePruner) getOrphanArchives(ctx context.Context) {
var archiveID string
// get all pkgs from kubernetes
for _, namespace := range utils.GetNamespaces() {
for _, namespace := range utils.DefaultNSResolver().FissionResourceNS {
pkgList, err := pruner.crdClient.CoreV1().Packages(namespace).List(ctx, metav1.ListOptions{})
if err != nil {
pruner.logger.Error("error getting package list from kubernetes", zap.Error(err))
pkg/utils/informer.go
+3 -28
View File
@@ -1,8 +1,6 @@
package utils
import (
"os"
"strings"
"time"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -13,37 +11,14 @@ import (
"k8s.io/client-go/tools/cache"
metricsapi "k8s.io/metrics/pkg/apis/metrics"
v1 "github.com/fission/fission/pkg/apis/core/v1"
fv1 "github.com/fission/fission/pkg/apis/core/v1"
"github.com/fission/fission/pkg/generated/clientset/versioned"
genInformer "github.com/fission/fission/pkg/generated/informers/externalversions"
)
const additionalNamespaces string = "FISSION_RESOURCE_NAMESPACES"
func GetNamespaces() []string {
envValue := os.Getenv(additionalNamespaces)
if len(envValue) == 0 {
return []string{
metav1.NamespaceAll,
}
}
informerNS := make([]string, 0)
lstNamespaces := strings.Split(envValue, ",")
for _, namespace := range lstNamespaces {
//check to handle string with additional comma at the end of string. eg- ns1,ns2,
if namespace != "" {
informerNS = append(informerNS, namespace)
}
}
return informerNS
}
func GetInformersForNamespaces(client versioned.Interface, defaultSync time.Duration, kind string) map[string]cache.SharedIndexInformer {
informers := make(map[string]cache.SharedIndexInformer)
for _, ns := range GetNamespaces() {
for _, ns := range DefaultNSResolver().FissionResourceNS {
factory := genInformer.NewFilteredSharedInformerFactory(client, defaultSync, ns, nil).Core().V1()
switch kind {
case fv1.CanaryConfigResource:
@@ -79,8 +54,8 @@ func GetInformerFactoryByReadyPod(client kubernetes.Interface, namespace string,
return informerFactory, nil
}
func GetInformerFactoryByExecutor(client kubernetes.Interface, executorType v1.ExecutorType, defaultResync time.Duration) (k8sInformers.SharedInformerFactory, error) {
executorLabel, err := labels.NewRequirement(v1.EXECUTOR_TYPE, selection.DoubleEquals, []string{string(executorType)})
func GetInformerFactoryByExecutor(client kubernetes.Interface, executorType fv1.ExecutorType, defaultResync time.Duration) (k8sInformers.SharedInformerFactory, error) {
executorLabel, err := labels.NewRequirement(fv1.EXECUTOR_TYPE, selection.DoubleEquals, []string{string(executorType)})
if err != nil {
return nil, err
}
pkg/utils/namespace.go
+98 -1
View File
@@ -2,22 +2,39 @@ package utils
import (
"os"
"strings"
"go.uber.org/zap"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/fission/fission/pkg/utils/loggerfactory"
)
const (
ENV_FUNCTION_NAMESPACE string = "FISSION_FUNCTION_NAMESPACE"
ENV_BUILDER_NAMESPACE string = "FISSION_BUILDER_NAMESPACE"
ENV_DEFAULT_NAMESPACE string = "FISSION_DEFAULT_NAMESPACE"
ENV_ADDITIONAL_NAMESPACE string = "FISSION_RESOURCE_NAMESPACES"
)
type NamespaceResolver struct {
type (
NamespaceResolver struct {
FunctionNamespace string
BuiderNamespace string
DefaultNamespace string
FissionResourceNS map[string]string
Logger *zap.Logger
}
options struct {
functionNS bool
builderNS bool
defaultNs bool
}
option func(options *options) *options
)
var nsResolver *NamespaceResolver
func init() {
@@ -25,7 +42,87 @@ func init() {
FunctionNamespace: os.Getenv(ENV_FUNCTION_NAMESPACE),
BuiderNamespace: os.Getenv(ENV_BUILDER_NAMESPACE),
DefaultNamespace: os.Getenv(ENV_DEFAULT_NAMESPACE),
FissionResourceNS: getNamespaces(),
Logger: loggerfactory.GetLogger(),
}
nsResolver.Logger.Debug("namespaces", zap.String("function_namespace", nsResolver.FunctionNamespace),
zap.String("builder_namespace", nsResolver.BuiderNamespace),
zap.String("default_namespace", nsResolver.DefaultNamespace),
zap.Any("fission_resource_namespace", listNamespaces(nsResolver.FissionResourceNS)))
}
// listNamespaces => convert namespaces from map to slice
func listNamespaces(namespaces map[string]string) []string {
ns := make([]string, 0)
for _, namespace := range namespaces {
ns = append(ns, namespace)
}
return ns
}
func WithBuilderNs() option {
return func(options *options) *options {
options.builderNS = true
return options
}
}
func WithFunctionNs() option {
return func(options *options) *options {
options.functionNS = true
return options
}
}
func WithDefaultNs() option {
return func(options *options) *options {
options.defaultNs = true
return options
}
}
func (nsr *NamespaceResolver) FissionNSWithOptions(option ...option) map[string]string {
var options options
for _, opt := range option {
options = *opt(&options)
}
fissionResourceNS := make(map[string]string)
for k, v := range nsr.FissionResourceNS {
fissionResourceNS[k] = v
}
if options.functionNS && nsr.FunctionNamespace != "" {
fissionResourceNS[nsr.FunctionNamespace] = nsr.FunctionNamespace
}
if options.builderNS && nsr.BuiderNamespace != "" {
fissionResourceNS[nsr.BuiderNamespace] = nsr.BuiderNamespace
}
if options.defaultNs && nsr.DefaultNamespace != "" {
fissionResourceNS[nsr.DefaultNamespace] = nsr.DefaultNamespace
}
nsr.Logger.Debug("fission resource namespaces", zap.Any("namespaces", listNamespaces(fissionResourceNS)))
return fissionResourceNS
}
func getNamespaces() map[string]string {
envValue := os.Getenv(ENV_ADDITIONAL_NAMESPACE)
if len(envValue) == 0 {
return map[string]string{
metav1.NamespaceDefault: metav1.NamespaceDefault,
}
}
lstNamespaces := strings.Split(envValue, ",")
namespaces := make(map[string]string, len(lstNamespaces))
for _, namespace := range lstNamespaces {
//check to handle string with additional comma at the end of string. eg- ns1,ns2,
if namespace != "" {
namespaces[namespace] = namespace
}
}
return namespaces
}
func (nsr *NamespaceResolver) GetBuilderNS(namespace string) string {
pkg/utils/namespace_test.go
+3 -1
View File
@@ -1,6 +1,8 @@
package utils
import "testing"
import (
"testing"
)
func TestNamespaceResolver(t *testing.T) {
t.Run("GetBuilderNS", func(t *testing.T) {