update chart version to v1.19.0-rc1 (#2761)

Signed-off-by: Shubham Bansal <shubhambansaliimtgn@gmail.com>

.github/workflows/lint-dashboards.yaml

@@ -24,7 +24,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v3
         with:
-          go-version: 1.18.5
+          go-version-file: "go.mod"
 
       - name: Install dashboard linter
         run: |

.github/workflows/lint.yaml

@@ -18,7 +18,7 @@ on:
   workflow_dispatch:
 
 env:
-  GOLANGCI_LINT_VERSION: v1.50.1
+  GOLANGCI_LINT_VERSION: v1.51.1
   GOLANGCI_LINT_TIMEOUT: 5m
 
 jobs:

.github/workflows/release.yaml

@@ -43,7 +43,14 @@ jobs:
       - name: Set up QEMU
         uses: docker/setup-qemu-action@v2
 
-      - name: Docker Login
+      - name: Login to ghcr.io
+        uses: docker/login-action@v2
+        with:
+          registry: ghcr.io
+          username: ${{ github.repository_owner }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Login to docker.io
         uses: docker/login-action@v2
         with:
           username: ${{ secrets.DOCKERHUB_USERNAME }}

.goreleaser.yml

@@ -84,8 +84,11 @@ dockers:
     image_templates:
       - "fission/builder:latest-amd64"
       - "fission/builder:{{ .Tag }}-amd64"
+      - "ghcr.io/fission/builder:latest-amd64"
+      - "ghcr.io/fission/builder:{{ .Tag }}-amd64"
     dockerfile: cmd/builder/Dockerfile
     build_flag_templates:
+      - "--label=org.opencontainers.image.source={{.GitURL}}"
       - "--platform=linux/amd64"
       - "--label=org.opencontainers.image.created={{.Date}}"
       - "--label=org.opencontainers.image.revision={{.FullCommit}}"
@@ -96,28 +99,60 @@ dockers:
     image_templates:
       - "fission/fetcher:latest-amd64"
       - "fission/fetcher:{{ .Tag }}-amd64"
+      - "ghcr.io/fission/fetcher:latest-amd64"
+      - "ghcr.io/fission/fetcher:{{ .Tag }}-amd64"
     dockerfile: cmd/fetcher/Dockerfile
+    build_flag_templates:
+      - "--label=org.opencontainers.image.source={{.GitURL}}"
+      - "--platform=linux/amd64"
+      - "--label=org.opencontainers.image.created={{.Date}}"
+      - "--label=org.opencontainers.image.revision={{.FullCommit}}"
+      - "--label=org.opencontainers.image.version={{.Tag}}"
   - <<: *docker-amd64
     ids:
       - fission-bundle
     image_templates:
       - "fission/fission-bundle:latest-amd64"
       - "fission/fission-bundle:{{ .Tag }}-amd64"
+      - "ghcr.io/fission/fission-bundle:latest-amd64"
+      - "ghcr.io/fission/fission-bundle:{{ .Tag }}-amd64"
     dockerfile: cmd/fission-bundle/Dockerfile
+    build_flag_templates:
+      - "--label=org.opencontainers.image.source={{.GitURL}}"
+      - "--platform=linux/amd64"
+      - "--label=org.opencontainers.image.created={{.Date}}"
+      - "--label=org.opencontainers.image.revision={{.FullCommit}}"
+      - "--label=org.opencontainers.image.version={{.Tag}}"
   - <<: *docker-amd64
     ids:
       - pre-upgrade-checks
     image_templates:
       - "fission/pre-upgrade-checks:latest-amd64"
       - "fission/pre-upgrade-checks:{{ .Tag }}-amd64"
+      - "ghcr.io/fission/pre-upgrade-checks:latest-amd64"
+      - "ghcr.io/fission/pre-upgrade-checks:{{ .Tag }}-amd64"
     dockerfile: cmd/preupgradechecks/Dockerfile
+    build_flag_templates:
+      - "--label=org.opencontainers.image.source={{.GitURL}}"
+      - "--platform=linux/amd64"
+      - "--label=org.opencontainers.image.created={{.Date}}"
+      - "--label=org.opencontainers.image.revision={{.FullCommit}}"
+      - "--label=org.opencontainers.image.version={{.Tag}}"
   - <<: *docker-amd64
     ids:
       - reporter
     image_templates:
       - "fission/reporter:latest-amd64"
       - "fission/reporter:{{ .Tag }}-amd64"
+      - "ghcr.io/fission/reporter:latest-amd64"
+      - "ghcr.io/fission/reporter:{{ .Tag }}-amd64"
     dockerfile: cmd/reporter/Dockerfile
+    build_flag_templates:
+      - "--label=org.opencontainers.image.source={{.GitURL}}"
+      - "--platform=linux/amd64"
+      - "--label=org.opencontainers.image.created={{.Date}}"
+      - "--label=org.opencontainers.image.revision={{.FullCommit}}"
+      - "--label=org.opencontainers.image.version={{.Tag}}"
   - &docker-arm64
     use: buildx
     goos: linux
@@ -127,8 +162,11 @@ dockers:
     image_templates:
       - "fission/builder:latest-arm64"
       - "fission/builder:{{ .Tag }}-arm64"
+      - "ghcr.io/fission/builder:latest-arm64"
+      - "ghcr.io/fission/builder:{{ .Tag }}-arm64"
     dockerfile: cmd/builder/Dockerfile
     build_flag_templates:
+      - "--label=org.opencontainers.image.source={{.GitURL}}"
       - "--platform=linux/arm64"
       - "--label=org.opencontainers.image.created={{.Date}}"
       - "--label=org.opencontainers.image.revision={{.FullCommit}}"
@@ -139,28 +177,60 @@ dockers:
     image_templates:
       - "fission/fetcher:latest-arm64"
       - "fission/fetcher:{{ .Tag }}-arm64"
+      - "ghcr.io/fission/fetcher:latest-arm64"
+      - "ghcr.io/fission/fetcher:{{ .Tag }}-arm64"
     dockerfile: cmd/fetcher/Dockerfile
+    build_flag_templates:
+      - "--label=org.opencontainers.image.source={{.GitURL}}"
+      - "--platform=linux/arm64"
+      - "--label=org.opencontainers.image.created={{.Date}}"
+      - "--label=org.opencontainers.image.revision={{.FullCommit}}"
+      - "--label=org.opencontainers.image.version={{.Tag}}"
   - <<: *docker-arm64
     ids:
       - fission-bundle
     image_templates:
       - "fission/fission-bundle:latest-arm64"
       - "fission/fission-bundle:{{ .Tag }}-arm64"
+      - "ghcr.io/fission/fission-bundle:latest-arm64"
+      - "ghcr.io/fission/fission-bundle:{{ .Tag }}-arm64"
     dockerfile: cmd/fission-bundle/Dockerfile
+    build_flag_templates:
+      - "--label=org.opencontainers.image.source={{.GitURL}}"
+      - "--platform=linux/arm64"
+      - "--label=org.opencontainers.image.created={{.Date}}"
+      - "--label=org.opencontainers.image.revision={{.FullCommit}}"
+      - "--label=org.opencontainers.image.version={{.Tag}}"
   - <<: *docker-arm64
     ids:
       - pre-upgrade-checks
     image_templates:
       - "fission/pre-upgrade-checks:latest-arm64"
       - "fission/pre-upgrade-checks:{{ .Tag }}-arm64"
+      - "ghcr.io/fission/pre-upgrade-checks:latest-arm64"
+      - "ghcr.io/fission/pre-upgrade-checks:{{ .Tag }}-arm64"
     dockerfile: cmd/preupgradechecks/Dockerfile
+    build_flag_templates:
+      - "--label=org.opencontainers.image.source={{.GitURL}}"
+      - "--platform=linux/arm64"
+      - "--label=org.opencontainers.image.created={{.Date}}"
+      - "--label=org.opencontainers.image.revision={{.FullCommit}}"
+      - "--label=org.opencontainers.image.version={{.Tag}}"
   - <<: *docker-arm64
     ids:
       - reporter
     image_templates:
       - "fission/reporter:latest-arm64"
       - "fission/reporter:{{ .Tag }}-arm64"
+      - "ghcr.io/fission/reporter:latest-arm64"
+      - "ghcr.io/fission/reporter:{{ .Tag }}-arm64"
     dockerfile: cmd/reporter/Dockerfile
+    build_flag_templates:
+      - "--label=org.opencontainers.image.source={{.GitURL}}"
+      - "--platform=linux/arm64"
+      - "--label=org.opencontainers.image.created={{.Date}}"
+      - "--label=org.opencontainers.image.revision={{.FullCommit}}"
+      - "--label=org.opencontainers.image.version={{.Tag}}"
   - &docker-armv7
     use: buildx
     goos: linux
@@ -171,8 +241,11 @@ dockers:
     image_templates:
       - "fission/builder:latest-armv7"
       - "fission/builder:{{ .Tag }}-armv7"
+      - "ghcr.io/fission/builder:latest-armv7"
+      - "ghcr.io/fission/builder:{{ .Tag }}-armv7"
     dockerfile: cmd/builder/Dockerfile
     build_flag_templates:
+      - "--label=org.opencontainers.image.source={{.GitURL}}"
       - "--platform=linux/arm/v7"
       - "--label=org.opencontainers.image.created={{.Date}}"
       - "--label=org.opencontainers.image.revision={{.FullCommit}}"
@@ -183,74 +256,156 @@ dockers:
     image_templates:
       - "fission/fetcher:latest-armv7"
       - "fission/fetcher:{{ .Tag }}-armv7"
+      - "ghcr.io/fission/fetcher:latest-armv7"
+      - "ghcr.io/fission/fetcher:{{ .Tag }}-armv7"
     dockerfile: cmd/fetcher/Dockerfile
+    build_flag_templates:
+      - "--label=org.opencontainers.image.source={{.GitURL}}"
+      - "--platform=linux/arm/v7"
+      - "--label=org.opencontainers.image.created={{.Date}}"
+      - "--label=org.opencontainers.image.revision={{.FullCommit}}"
+      - "--label=org.opencontainers.image.version={{.Tag}}"
   - <<: *docker-armv7
     ids:
       - fission-bundle
     image_templates:
       - "fission/fission-bundle:latest-armv7"
       - "fission/fission-bundle:{{ .Tag }}-armv7"
+      - "ghcr.io/fission/fission-bundle:latest-armv7"
+      - "ghcr.io/fission/fission-bundle:{{ .Tag }}-armv7"
     dockerfile: cmd/fission-bundle/Dockerfile
+    build_flag_templates:
+      - "--label=org.opencontainers.image.source={{.GitURL}}"
+      - "--platform=linux/arm/v7"
+      - "--label=org.opencontainers.image.created={{.Date}}"
+      - "--label=org.opencontainers.image.revision={{.FullCommit}}"
+      - "--label=org.opencontainers.image.version={{.Tag}}"
   - <<: *docker-armv7
     ids:
       - pre-upgrade-checks
     image_templates:
       - "fission/pre-upgrade-checks:latest-armv7"
       - "fission/pre-upgrade-checks:{{ .Tag }}-armv7"
+      - "ghcr.io/fission/pre-upgrade-checks:latest-armv7"
+      - "ghcr.io/fission/pre-upgrade-checks:{{ .Tag }}-armv7"
     dockerfile: cmd/preupgradechecks/Dockerfile
+    build_flag_templates:
+      - "--label=org.opencontainers.image.source={{.GitURL}}"
+      - "--platform=linux/arm/v7"
+      - "--label=org.opencontainers.image.created={{.Date}}"
+      - "--label=org.opencontainers.image.revision={{.FullCommit}}"
+      - "--label=org.opencontainers.image.version={{.Tag}}"
   - <<: *docker-armv7
     ids:
       - reporter
     image_templates:
       - "fission/reporter:latest-armv7"
       - "fission/reporter:{{ .Tag }}-armv7"
+      - "ghcr.io/fission/reporter:latest-armv7"
+      - "ghcr.io/fission/reporter:{{ .Tag }}-armv7"
     dockerfile: cmd/reporter/Dockerfile
+    build_flag_templates:
+      - "--label=org.opencontainers.image.source={{.GitURL}}"
+      - "--platform=linux/arm/v7"
+      - "--label=org.opencontainers.image.created={{.Date}}"
+      - "--label=org.opencontainers.image.revision={{.FullCommit}}"
+      - "--label=org.opencontainers.image.version={{.Tag}}"
 docker_manifests:
+  - name_template: ghcr.io/fission/builder:{{ .Tag }}
+    image_templates:
+      - ghcr.io/fission/builder:{{ .Tag }}-amd64
+      - ghcr.io/fission/builder:{{ .Tag }}-arm64
+      - ghcr.io/fission/builder:{{ .Tag }}-armv7
   - name_template: fission/builder:{{ .Tag }}
     image_templates:
       - fission/builder:{{ .Tag }}-amd64
       - fission/builder:{{ .Tag }}-arm64
       - fission/builder:{{ .Tag }}-armv7
+  - name_template: ghcr.io/fission/fetcher:{{ .Tag }}
+    image_templates:
+      - ghcr.io/fission/fetcher:{{ .Tag }}-amd64
+      - ghcr.io/fission/fetcher:{{ .Tag }}-arm64
+      - ghcr.io/fission/fetcher:{{ .Tag }}-armv7
   - name_template: fission/fetcher:{{ .Tag }}
     image_templates:
       - fission/fetcher:{{ .Tag }}-amd64
       - fission/fetcher:{{ .Tag }}-arm64
       - fission/fetcher:{{ .Tag }}-armv7
+  - name_template: ghcr.io/fission/fission-bundle:{{ .Tag }}
+    image_templates:
+      - ghcr.io/fission/fission-bundle:{{ .Tag }}-amd64
+      - ghcr.io/fission/fission-bundle:{{ .Tag }}-arm64
+      - ghcr.io/fission/fission-bundle:{{ .Tag }}-armv7
   - name_template: fission/fission-bundle:{{ .Tag }}
     image_templates:
       - fission/fission-bundle:{{ .Tag }}-amd64
       - fission/fission-bundle:{{ .Tag }}-arm64
       - fission/fission-bundle:{{ .Tag }}-armv7
+  - name_template: ghcr.io/fission/pre-upgrade-checks:{{ .Tag }}
+    image_templates:
+      - ghcr.io/fission/pre-upgrade-checks:{{ .Tag }}-amd64
+      - ghcr.io/fission/pre-upgrade-checks:{{ .Tag }}-arm64
+      - ghcr.io/fission/pre-upgrade-checks:{{ .Tag }}-armv7
   - name_template: fission/pre-upgrade-checks:{{ .Tag }}
     image_templates:
       - fission/pre-upgrade-checks:{{ .Tag }}-amd64
       - fission/pre-upgrade-checks:{{ .Tag }}-arm64
       - fission/pre-upgrade-checks:{{ .Tag }}-armv7
+  - name_template: ghcr.io/fission/reporter:{{ .Tag }}
+    image_templates:
+      - ghcr.io/fission/reporter:{{ .Tag }}-amd64
+      - ghcr.io/fission/reporter:{{ .Tag }}-arm64
+      - ghcr.io/fission/reporter:{{ .Tag }}-armv7
   - name_template: fission/reporter:{{ .Tag }}
     image_templates:
       - fission/reporter:{{ .Tag }}-amd64
       - fission/reporter:{{ .Tag }}-arm64
       - fission/reporter:{{ .Tag }}-armv7
+  - name_template: ghcr.io/fission/builder:latest
+    image_templates:
+      - ghcr.io/fission/builder:latest-amd64
+      - ghcr.io/fission/builder:latest-arm64
+      - ghcr.io/fission/builder:latest-armv7
   - name_template: fission/builder:latest
     image_templates:
       - fission/builder:latest-amd64
       - fission/builder:latest-arm64
       - fission/builder:latest-armv7
+  - name_template: ghcr.io/fission/fetcher:latest
+    image_templates:
+      - ghcr.io/fission/fetcher:latest-amd64
+      - ghcr.io/fission/fetcher:latest-arm64
+      - ghcr.io/fission/fetcher:latest-armv7
   - name_template: fission/fetcher:latest
     image_templates:
       - fission/fetcher:latest-amd64
       - fission/fetcher:latest-arm64
-      - fission/fetcher:latest-armv7
+      - fission/fetcher:latest-armv7 
+  - name_template: ghcr.io/fission/fission-bundle:latest
+    image_templates:
+      - ghcr.io/fission/fission-bundle:latest-amd64
+      - ghcr.io/fission/fission-bundle:latest-arm64
+      - ghcr.io/fission/fission-bundle:latest-armv7
   - name_template: fission/fission-bundle:latest
     image_templates:
       - fission/fission-bundle:latest-amd64
       - fission/fission-bundle:latest-arm64
       - fission/fission-bundle:latest-armv7
+  - name_template: ghcr.io/fission/pre-upgrade-checks:latest
+    image_templates:
+      - ghcr.io/fission/pre-upgrade-checks:latest-amd64
+      - ghcr.io/fission/pre-upgrade-checks:latest-arm64
+      - ghcr.io/fission/pre-upgrade-checks:latest-armv7
   - name_template: fission/pre-upgrade-checks:latest
     image_templates:
       - fission/pre-upgrade-checks:latest-amd64
       - fission/pre-upgrade-checks:latest-arm64
       - fission/pre-upgrade-checks:latest-armv7
+  - name_template: ghcr.io/fission/reporter:latest
+    image_templates:
+      - ghcr.io/fission/reporter:latest-amd64
+      - ghcr.io/fission/reporter:latest-arm64
+      - ghcr.io/fission/reporter:latest-armv7
   - name_template: fission/reporter:latest
     image_templates:
       - fission/reporter:latest-amd64

charts/fission-all/Chart.yaml

@@ -1,7 +1,7 @@
 apiVersion: v2
 name: fission-all
-version: v1.18.0
+version: v1.19.0-rc1
-appVersion: v1.18.0
+appVersion: v1.19.0-rc1
 description: Fission is a fast serverless framework for Kubernetes.
 home: https://fission.io/
 icon: https://fission.io/images/fission-logo-white.svg

charts/fission-all/templates/_fission-kubernetes-roles.tpl

@@ -12,6 +12,15 @@ rules:
   - list
   - watch
   - patch
+- apiGroups:
+  - ""
+  resources:
+  - configmaps
+  - secrets
+  verbs:
+  - get
+  - list
+  - watch
 - apiGroups:
   - apps
   resources:

charts/fission-all/templates/_helpers.tpl

@@ -55,6 +55,22 @@ This template generates the image name for the deployment depending on the value
 {{- end }}
 {{- end -}}
 
+{{- define "reporterImage" -}}
+{{- if .Values.repository -}}
+  {{- if eq .Values.imageTag "" -}}
+    {{ .Values.repository }}/{{ .Values.postInstallReportImage }}
+  {{- else -}}
+    {{ .Values.repository }}/{{ .Values.postInstallReportImage }}:{{ .Values.imageTag }}
+  {{- end }}
+{{- else -}}
+  {{- if eq .Values.imageTag "" -}}
+    {{ .Values.postInstallReportImage }}
+  {{- else -}}
+    {{ .Values.postInstallReportImage }}:{{ .Values.imageTag }}
+  {{- end }}
+{{- end }}
+{{- end -}}
+
 {{- define "opentelemtry.envs" }}
 - name: OTEL_EXPORTER_OTLP_ENDPOINT
   value: "{{ .Values.openTelemetry.otlpCollectorEndpoint }}"

charts/fission-all/templates/analytics/nonhelm-install-job.yaml

@@ -26,11 +26,7 @@ spec:
       restartPolicy: Never
       containers:
         - name: post-install-job
-        {{- if .Values.imageTag }}
+          image: {{ include "reporterImage" . | quote }}
-          image: {{ .Values.postInstallReportImage }}:{{ .Values.imageTag }}
-        {{- else }}
-          image: {{ .Values.postInstallReportImage }}
-        {{- end }}
           imagePullPolicy: {{ .Values.pullPolicy }}
           command: [ "/reporter" ]
           args: ["event", "-c", "fission-use", "-a", "yaml-post-install", "-l", "{{ .Chart.Name }}-{{ .Chart.Version }}"]

charts/fission-all/templates/analytics/post-install-job.yaml

@@ -30,11 +30,7 @@ spec:
       restartPolicy: Never
       containers:
         - name: post-install-job
-        {{- if .Values.imageTag }}
+          image: {{ include "reporterImage" . | quote }}
-          image: {{ .Values.postInstallReportImage }}:{{ .Values.imageTag }}
-        {{- else }}
-          image: {{ .Values.postInstallReportImage }}
-        {{- end }}
           imagePullPolicy: {{ .Values.pullPolicy }}
           command: [ "/reporter" ]
           args: ["event", "-c", "fission-use", "-a", "helm-post-install", "-l", "{{ .Chart.Name }}-{{ .Chart.Version }}"]

charts/fission-all/templates/analytics/post-upgrade-job.yaml

@@ -30,11 +30,7 @@ spec:
       restartPolicy: Never
       containers:
         - name: post-upgrade-job
-        {{- if .Values.imageTag }}
+          image: {{ include "reporterImage" . | quote }}
-          image: {{ .Values.postInstallReportImage }}:{{ .Values.imageTag }}
-        {{- else }}
-          image: {{ .Values.postInstallReportImage }}
-        {{- end }}
           imagePullPolicy: {{ .Values.pullPolicy }}
           command: [ "/reporter" ]
           args: ["event", "-c", "fission-use", "-a", "helm-post-upgrade", "-l", "{{ .Chart.Name }}-{{ .Chart.Version }}"]

charts/fission-all/templates/buildermgr/role-fission-cr.yaml

@@ -5,3 +5,9 @@
 {{ include "fission-role-generator" (merge (dict "namespace" $namespace "component" "buildermgr") $) }}
 {{- end }}
 {{- end }}
+{{- if .Values.builderNamespace -}}
+{{ include "fission-role-generator" (merge (dict "namespace" .Values.builderNamespace "component" "buildermgr") $) }}
+{{- end }}
+{{- if .Values.functionNamespace -}}
+{{ include "fission-role-generator" (merge (dict "namespace" .Values.functionNamespace "component" "buildermgr") $) }}
+{{- end }}

charts/fission-all/templates/executor/role-fission-cr.yaml

@@ -5,3 +5,9 @@
 {{ include "fission-role-generator" (merge (dict "namespace" $namespace "component" "executor") $) }}
 {{- end }}
 {{- end }}
+{{- if .Values.builderNamespace -}}
+{{ include "fission-role-generator" (merge (dict "namespace" .Values.builderNamespace "component" "executor") $) }}
+{{- end }}
+{{- if .Values.functionNamespace -}}
+{{ include "fission-role-generator" (merge (dict "namespace" .Values.functionNamespace "component" "executor") $) }}
+{{- end }}

charts/fission-all/templates/mqt-fission-kafka/deployment.yaml

@@ -25,11 +25,7 @@ spec:
     spec:
       containers:
       - name: mqtrigger
-      {{- if eq .Values.imageTag "" }}
+        image: {{ include "fission-bundleImage" . | quote }}
-        image: "{{ .Values.image }}"
-      {{- else }}
-        image: "{{ .Values.image }}:{{ .Values.imageTag }}"
-      {{- end }}
         imagePullPolicy: {{ .Values.pullPolicy }}
         command: ["/fission-bundle"]
         args: ["--mqt", "--routerUrl", "http://router.{{ .Release.Namespace }}"]

charts/fission-all/templates/router/role-kubernetes.yaml

@@ -1,4 +1,5 @@
 {{- include "kubernetes-role-generator" (merge (dict "namespace" .Values.defaultNamespace "component" "router") .) }}
+{{- include "kubernetes-role-generator" (merge (dict "namespace" .Release.Namespace "component" "router") .) }}
 
 {{- if gt (len .Values.additionalFissionNamespaces) 0 }}
 {{- range $namespace := $.Values.additionalFissionNamespaces }}

charts/fission-all/templates/router/secret.yaml

@@ -6,7 +6,7 @@ metadata:
   labels:
     chart: "{{ .Chart.Name }}-{{ .Chart.Version }}"
   annotations:
-    "helm.sh/hook": pre-install
+    "helm.sh/hook": pre-install,pre-upgrade
 data:
   username: {{ .Values.authentication.authUsername | b64enc | quote }}
   password: {{ randAlphaNum 20 | b64enc | quote }}

charts/fission-all/values.yaml

@@ -14,7 +14,7 @@ routerServiceType: LoadBalancer
 ## repository represents base repository for images used in the chart.
 ## Keep it empty for using existing local image
 ##
-repository: index.docker.io
+repository: ghcr.io
 
 ## image represents the base image fission-bundle used by multiple Fission components.
 ## We alter arguments to the image to run a particular component.
@@ -25,7 +25,7 @@ image: fission/fission-bundle
 ## It is also used by the chart to identify version of the few more images apart from fission-bundle.
 ## Keep it empty for using latest tag.
 ##
-imageTag: v1.18.0
+imageTag: v1.19.0-rc1
 
 ## pullPolicy represents the pull policy to use for images in the chart.
 ##
@@ -63,8 +63,7 @@ routerPort: 31314
 
 ## defaultNamespace represents the namespace in which Fission custom resources will be created by the Fission user.
 ## This is different from the release namespace.
-## Please consider setting `singleDefaultNamespace` and `additionalFissionNamespaces` if you want
+## Please consider setting `additionalFissionNamespaces` if you want more than one namespace to be used for Fission custom resources.
-## more than one namespace to be used for Fission custom resources.
 ##
 defaultNamespace: default
 
@@ -104,7 +103,7 @@ fetcher:
   ## image represents the image of the fetcher component.
   image: fission/fetcher
   ## imageTag represents the tag of the image of the fetcher component.
-  imageTag: v1.18.0
+  imageTag: v1.19.0-rc1
 
   ## Fetcher is only for to downloading or uploading archive.
   ## Normally, you don't need to change the value here, unless necessary.
@@ -686,7 +685,7 @@ preUpgradeChecks:
   image: fission/pre-upgrade-checks
   ## pre-install/pre-upgrade checks image version
   ##
-  imageTag: v1.18.0
+  imageTag: v1.19.0-rc1
 
 ## Fission post-install/post-upgrade reporting live in this image
 ##

cmd/builder/Dockerfile

@@ -1,4 +1,4 @@
-FROM alpine:3.16
+FROM alpine:3.17
 RUN apk add --update ca-certificates
 COPY builder /builder
 ENTRYPOINT ["/builder"]

cmd/fetcher/Dockerfile

@@ -1,4 +1,4 @@
-FROM alpine:3.16
+FROM alpine:3.17
 RUN apk add --update ca-certificates
 COPY fetcher /
 ENTRYPOINT ["/fetcher"]

cmd/fission-bundle/Dockerfile

@@ -1,4 +1,4 @@
-FROM alpine:3.16
+FROM alpine:3.17
 RUN apk add --update ca-certificates
 COPY fission-bundle /
 ENTRYPOINT ["/fission-bundle"]

cmd/preupgradechecks/Dockerfile

@@ -1,4 +1,4 @@
-FROM alpine:3.16
+FROM alpine:3.17
 RUN apk add --update ca-certificates
 COPY pre-upgrade-checks /
 ENTRYPOINT ["/pre-upgrade-checks"]

cmd/reporter/Dockerfile

@@ -1,4 +1,4 @@
-FROM alpine:3.16
+FROM alpine:3.17
 RUN apk add --update ca-certificates
 COPY reporter /
 ENTRYPOINT ["/reporter"]

go.mod

@@ -1,6 +1,6 @@
 module github.com/fission/fission
 
-go 1.19
+go 1.20
 
 require (
 	github.com/Shopify/sarama v1.37.2
@@ -44,7 +44,7 @@ require (
 	go.opentelemetry.io/otel/sdk v1.11.2
 	go.opentelemetry.io/otel/trace v1.11.2
 	go.uber.org/zap v1.24.0
-	golang.org/x/net v0.4.0
+	golang.org/x/net v0.7.0
 	google.golang.org/grpc v1.51.0
 	k8s.io/api v0.25.4
 	k8s.io/apiextensions-apiserver v0.25.4
@@ -146,7 +146,7 @@ require (
 	github.com/nwaples/rardecode v1.1.0 // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/opencontainers/image-spec v1.0.2 // indirect
-	github.com/opencontainers/runc v1.1.2 // indirect
+	github.com/opencontainers/runc v1.1.5 // indirect
 	github.com/pierrec/lz4/v4 v4.1.17 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.3.0 // indirect
@@ -155,7 +155,7 @@ require (
 	github.com/rogpeppe/go-internal v1.9.0 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/sergi/go-diff v1.1.0 // indirect
-	github.com/sirupsen/logrus v1.8.1 // indirect
+	github.com/sirupsen/logrus v1.9.0 // indirect
 	github.com/ulikunitz/xz v0.5.9 // indirect
 	github.com/xanzy/ssh-agent v0.3.2 // indirect
 	github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 // indirect
@@ -169,12 +169,12 @@ require (
 	go.uber.org/atomic v1.9.0 // indirect
 	go.uber.org/multierr v1.8.0 // indirect
 	golang.org/x/crypto v0.3.0 // indirect
-	golang.org/x/image v0.0.0-20190802002840-cff245a6509b // indirect
+	golang.org/x/image v0.5.0 // indirect
 	golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect
 	golang.org/x/oauth2 v0.0.0-20220411215720-9780585627b5 // indirect
-	golang.org/x/sys v0.3.0 // indirect
+	golang.org/x/sys v0.5.0 // indirect
-	golang.org/x/term v0.3.0 // indirect
+	golang.org/x/term v0.5.0 // indirect
-	golang.org/x/text v0.5.0 // indirect
+	golang.org/x/text v0.7.0 // indirect
 	golang.org/x/time v0.0.0-20220609170525-579cf78fd858 // indirect
 	golang.org/x/tools v0.1.12 // indirect
 	golang.org/x/xerrors v0.0.0-20220411194840-2f41105eb62f // indirect

go.sum

@@ -556,8 +556,8 @@ github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8
 github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
 github.com/opencontainers/image-spec v1.0.2 h1:9yCKha/T5XdGtO0q9Q9a6T5NUCsTn/DrBg0D7ufOcFM=
 github.com/opencontainers/image-spec v1.0.2/go.mod h1:BtxoFyWECRxE4U/7sNtV5W15zMzWCbyJoFRP3s7yZA0=
-github.com/opencontainers/runc v1.1.2 h1:2VSZwLx5k/BfsBxMMipG/LYUnmqOD/BPkIVgQUcTlLw=
+github.com/opencontainers/runc v1.1.5 h1:L44KXEpKmfWDcS02aeGm8QNTFXTo2D+8MYGDIJ/GDEs=
-github.com/opencontainers/runc v1.1.2/go.mod h1:Tj1hFw6eFWp/o33uxGf5yF2BX5yz2Z6iptFpuvbbKqc=
+github.com/opencontainers/runc v1.1.5/go.mod h1:1J5XiS+vdZ3wCyZybsuxXZWGrgSr8fFJHLXuG2PsnNg=
 github.com/opencontainers/runtime-spec v1.0.3-0.20210326190908-1c3f411f0417/go.mod h1:jwyrGlmzljRJv/Fgzds9SsS/C5hL+LL3ko9hs6T5lQ0=
 github.com/opencontainers/selinux v1.10.0/go.mod h1:2i0OySw99QjzBBQByd1Gr9gSjvuho1lHsJxIJ3gGbJI=
 github.com/ory/dockertest v3.3.5+incompatible h1:iLLK6SQwIhcbrG783Dghaaa3WPzGc+4Emza6EbVUUGA=
@@ -622,7 +622,7 @@ github.com/sabhiram/go-gitignore v0.0.0-20210923224102-525f6e181f06/go.mod h1:+e
 github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
 github.com/satori/go.uuid v1.2.1-0.20181028125025-b2ce2384e17b h1:gQZ0qzfKHQIybLANtM3mBXNUtOfsCFXeTsnBqCsx1KM=
 github.com/satori/go.uuid v1.2.1-0.20181028125025-b2ce2384e17b/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
-github.com/seccomp/libseccomp-golang v0.9.2-0.20210429002308-3879420cc921/go.mod h1:JA8cRccbGaA1s33RQf7Y1+q9gHmZX1yB/z9WDN1C6fg=
+github.com/seccomp/libseccomp-golang v0.9.2-0.20220502022130-f33da4d89646/go.mod h1:JA8cRccbGaA1s33RQf7Y1+q9gHmZX1yB/z9WDN1C6fg=
 github.com/sergi/go-diff v1.1.0 h1:we8PVUC3FE2uYfodKH/nBHMSetSfHDR6scGdBi+erh0=
 github.com/sergi/go-diff v1.1.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM=
 github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
@@ -631,8 +631,9 @@ github.com/sirupsen/logrus v1.4.1/go.mod h1:ni0Sbl8bgC9z8RoU9G6nDWqqs/fq4eDPysMB
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88=
 github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
-github.com/sirupsen/logrus v1.8.1 h1:dJKuHgqk1NNQlqoA6BTlM1Wf9DOH3NBjQyu0h9+AZZE=
 github.com/sirupsen/logrus v1.8.1/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
+github.com/sirupsen/logrus v1.9.0 h1:trlNQbNUG3OdDrDil03MCb1H2o9nJ1x4/5LYw7byDE0=
+github.com/sirupsen/logrus v1.9.0/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
 github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
@@ -685,6 +686,7 @@ github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9de
 github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
+github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
 go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8=
@@ -763,8 +765,9 @@ golang.org/x/exp v0.0.0-20200119233911-0405dc783f0a/go.mod h1:2RIsYlXP63K8oxa1u0
 golang.org/x/exp v0.0.0-20200207192155-f17229e696bd/go.mod h1:J/WKrq2StrnmMY6+EHIKF9dgMWnmCNThgcyBT1FY9mM=
 golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6/go.mod h1:3jZMyOhIsHpP37uCMkUooju7aAi5cS1Q23tOzKc+0MU=
 golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js=
-golang.org/x/image v0.0.0-20190802002840-cff245a6509b h1:+qEpEAPhDZ1o0x3tHzZTQDArnOixOzGD9HUJfcg0mb4=
 golang.org/x/image v0.0.0-20190802002840-cff245a6509b/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
+golang.org/x/image v0.5.0 h1:5JMiNunQeQw++mMOz48/ISeNu3Iweh/JaZU8ZLqHRrI=
+golang.org/x/image v0.5.0/go.mod h1:FVC7BI/5Ym8R25iw5OLsgshdUBbT1h5jZTpA+mvAdZ4=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
 golang.org/x/lint v0.0.0-20190301231843-5614ed5bae6f/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
@@ -838,9 +841,10 @@ golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qx
 golang.org/x/net v0.0.0-20211216030914-fe4d6282115f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
 golang.org/x/net v0.0.0-20220225172249-27dd8689420f/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
+golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.0.0-20220725212005-46097bf591d3/go.mod h1:AaygXjzTFtRAg2ttMY5RMuhpJ3cNnI0XpyFJD1iQRSM=
-golang.org/x/net v0.4.0 h1:Q5QPcMlvfxFTAPV0+07Xz/MpK9NTXu2VDUuy0FeMfaU=
+golang.org/x/net v0.7.0 h1:rJrUqqhjsgNp7KqAIc25s9pZnjU7TUcSY7HcVZjdn1g=
-golang.org/x/net v0.4.0/go.mod h1:MBQ8lrhLObU/6UmLb4fmbmk5OcyYmqtbGd/9yIeKjEE=
+golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -871,6 +875,7 @@ golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220923202941-7f9b1623fab7 h1:ZrnxWX62AgTKOSagEqxvb3ffipvEDX2pl7E1TdqLqIc=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -947,15 +952,16 @@ golang.org/x/sys v0.0.0-20211116061358-0a5406a5449c/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20211124211545-fe61309f8881/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.3.0 h1:w8ZOecv6NaNa/zC8944JTU3vz4u6Lagfk4RPQxv92NQ=
+golang.org/x/sys v0.5.0 h1:MUK/U/4lj1t1oPg0HfuXDN/Z1wv31ZJ/YcPiGccS4DU=
-golang.org/x/sys v0.3.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
-golang.org/x/term v0.3.0 h1:qoo4akIqOcDME5bhc/NgxUdovd6BSS2uMsVjB56q1xI=
+golang.org/x/term v0.5.0 h1:n2a8QNdAb0sZNpU9R1ALUXBbY+w51fCQDN+7EdxNBsY=
-golang.org/x/term v0.3.0/go.mod h1:q750SLmJuPmVoN1blW3UFBPREJfb1KmY3vwxfr+nFDA=
+golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -965,8 +971,8 @@ golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
-golang.org/x/text v0.5.0 h1:OLmvp0KP+FVG99Ct/qFiL/Fhk4zp4QQnZ7b2U+5piUM=
+golang.org/x/text v0.7.0 h1:4BRB4x83lYWy72KwLD/qYDuTu7q9PjSagHvijDw7cLo=
-golang.org/x/text v0.5.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=

pkg/apis/core/v1/types.go

@@ -22,6 +22,11 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+const (
+	DefaultConcurrency    = 500
+	DefaultRequestsPerPod = 1
+)
+
 //
 // To add a Fission CRD type:
 //   1. Create a "spec" type, for everything in the type except metadata
@@ -863,3 +868,17 @@ type (
 func (a Archive) IsEmpty() bool {
 	return len(a.Literal) == 0 && len(a.URL) == 0
 }
+
+func (fn Function) GetConcurrency() int {
+	if fn.Spec.Concurrency == 0 {
+		return DefaultConcurrency
+	}
+	return fn.Spec.Concurrency
+}
+
+func (fn Function) GetRequestPerPod() int {
+	if fn.Spec.RequestsPerPod == 0 {
+		return DefaultRequestsPerPod
+	}
+	return fn.Spec.RequestsPerPod
+}

pkg/builder/client/client.go

@@ -21,10 +21,9 @@ import (
 	"context"
 	"encoding/json"
 	"io"
-	"net/http"
 	"strings"
-	"time"
 
+	"github.com/hashicorp/go-retryablehttp"
 	"github.com/pkg/errors"
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 	"go.uber.org/zap"
@@ -39,12 +38,13 @@ type (
 	Client struct {
 		logger     *zap.Logger
 		url        string
-		httpClient *http.Client
+		httpClient *retryablehttp.Client
 	}
 )
 
 func MakeClient(logger *zap.Logger, builderUrl string) *Client {
-	hc := &http.Client{Transport: otelhttp.NewTransport(http.DefaultTransport)}
+	hc := retryablehttp.NewClient()
+	hc.HTTPClient.Transport = otelhttp.NewTransport(hc.HTTPClient.Transport)
 	return &Client{
 		logger:     logger.Named("builder_client"),
 		url:        strings.TrimSuffix(builderUrl, "/"),
@@ -60,27 +60,10 @@ func (c *Client) Build(ctx context.Context, req *builder.PackageBuildRequest) (*
 		return nil, errors.Wrap(err, "error marshaling json")
 	}
 
-	maxRetries := 20
+	resp, err := ctxhttp.Post(ctx, c.httpClient.StandardClient(), c.url, "application/json", bytes.NewReader(body))
-	var resp *http.Response
+	if err != nil {
-
+		return nil, ferror.MakeErrorFromHTTP(resp)
-	for i := 0; i < maxRetries; i++ {
-		resp, err = ctxhttp.Post(ctx, c.httpClient, c.url, "application/json", bytes.NewReader(body))
-		if err == nil {
-			if resp.StatusCode == 200 {
-				break
-			}
-			err = ferror.MakeErrorFromHTTP(resp)
-		}
-
-		if i < maxRetries-1 {
-			time.Sleep(50 * time.Duration(2*i) * time.Millisecond)
-			logger.Error("error building package, retrying", zap.Error(err))
-			continue
-		}
-
-		return nil, err
 	}
-
 	defer resp.Body.Close()
 
 	rBody, err := io.ReadAll(resp.Body)

pkg/error/httperror.go

@@ -17,6 +17,7 @@ limitations under the License.
 package error
 
 import (
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -106,8 +107,8 @@ func (err Error) Description() string {
 func GetHTTPError(err error) (int, string) {
 	var msg string
 	var code int
-	fe, ok := err.(Error)
+	var fe Error
-	if ok {
+	if errors.As(err, &fe) {
 		code = fe.HTTPStatus()
 		msg = fe.Message
 	} else {
@@ -118,10 +119,11 @@ func GetHTTPError(err error) (int, string) {
 }
 
 func IsNotFound(err error) bool {
-	fe, ok := err.(Error)
+	var fe Error
-	if !ok {
+	if !errors.As(err, &fe) {
 		return false
 	}
+
 	return fe.Code == ErrorNotFound
 }
 

pkg/error/httperror_test.go

@@ -0,0 +1,36 @@
+package error
+
+import (
+	"net/http"
+	"testing"
+
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIsNotFound(t *testing.T) {
+	errs := map[error]bool{
+		nil: false,
+		MakeError(ErrorNotFound, "someone not found"):                                          true,
+		MakeError(ErrorTooManyRequests, "too many requests"):                                   false,
+		errors.Wrap(MakeError(ErrorNotFound, "someone not found"), "other information"):        true,
+		errors.Wrap(MakeError(ErrorTooManyRequests, "too many requests"), "other information"): false,
+	}
+
+	for err, want := range errs {
+		assert.Equal(t, want, IsNotFound(err))
+	}
+}
+
+func TestGetHTTPError(t *testing.T) {
+	errs := map[int]error{
+		http.StatusBadRequest:      MakeError(ErrorInvalidArgument, ""),
+		http.StatusConflict:        errors.Wrap(MakeError(ErrorNameExists, ""), ""),
+		http.StatusNotFound:        errors.Wrap(MakeError(ErrorNotFound, ""), ""),
+		http.StatusTooManyRequests: errors.Wrap(MakeError(ErrorTooManyRequests, "too many requests"), "other information"),
+	}
+	for want, err := range errs {
+		code, _ := GetHTTPError(err)
+		assert.Equal(t, want, code)
+	}
+}

pkg/executor/api.go

@@ -63,19 +63,10 @@ func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *htt
 		zap.String("function_name", fn.ObjectMeta.Name),
 		zap.String("function_namespace", fn.ObjectMeta.Namespace))
 	if t == fv1.ExecutorTypePoolmgr && !fn.Spec.OnceOnly {
-		concurrency := fn.Spec.Concurrency
+		fsvc, err := et.GetFuncSvcFromCache(ctx, fn)
-		if concurrency == 0 {
-			concurrency = 500
-		}
-		requestsPerpod := fn.Spec.RequestsPerPod
-		if requestsPerpod == 0 {
-			requestsPerpod = 1
-		}
-		fsvc, active, err := et.GetFuncSvcFromPoolCache(ctx, fn, requestsPerpod)
 		// check if its a cache hit (check if there is already specialized function pod that can serve another request)
 		if err == nil {
 			// if a pod is already serving request then it already exists else validated
-			logger.Debug("from cache", zap.Int("active", active))
 			if et.IsValid(ctx, fsvc) {
 				// Cached, return svc address
 				logger.Debug("served from cache", zap.String("name", fsvc.Name), zap.String("address", fsvc.Address))
@@ -87,15 +78,19 @@ func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *htt
 				zap.String("function_namespace", fn.ObjectMeta.Namespace),
 				zap.String("address", fsvc.Address))
 			et.DeleteFuncSvcFromCache(ctx, fsvc)
-			active--
+		} else {
+			code, msg := ferror.GetHTTPError(err)
+			if code == http.StatusNotFound {
+				logger.Debug("cache miss", zap.String("function_name", fn.ObjectMeta.Name))
+			} else {
+				logger.Error("error getting service for function",
+					zap.Error(err),
+					zap.String("function_name", fn.ObjectMeta.Name))
+				http.Error(w, msg, code)
+				return
+			}
 		}
 
-		if active >= concurrency {
-			errMsg := fmt.Sprintf("max concurrency reached for %v. All %v instance are active", fn.ObjectMeta.Name, concurrency)
-			logger.Error("error occurred", zap.String("error", errMsg))
-			http.Error(w, html.EscapeString(errMsg), http.StatusTooManyRequests)
-			return
-		}
 	} else if t == fv1.ExecutorTypeNewdeploy || t == fv1.ExecutorTypeContainer {
 		fsvc, err := et.GetFuncSvcFromCache(ctx, fn)
 		if err == nil {

pkg/executor/executor.go

@@ -60,7 +60,6 @@ type (
 		requestChan chan *createFuncServiceRequest
 		fsCreateWg  sync.Map
 	}
-
 	createFuncServiceRequest struct {
 		context  context.Context
 		function *fv1.Function

pkg/executor/executor_test.go

@@ -101,7 +101,6 @@ func createSvc(ctx context.Context, kubeClient kubernetes.Interface, ns string,
 func TestExecutor(t *testing.T) {
 	// run in a random namespace so we can have concurrent tests
 	// on a given cluster
-	rand.Seed(time.Now().UTC().UnixNano())
 	testID := rand.Intn(999)
 	fissionNs := fmt.Sprintf("test-%v", testID)
 	functionNs := fmt.Sprintf("test-function-%v", testID)

pkg/executor/executortype/container/containermgr.go

@@ -193,12 +193,6 @@ func (caaf *Container) DeleteFuncSvcFromCache(ctx context.Context, fsvc *fscache
 	caaf.fsCache.DeleteEntry(fsvc)
 }
 
-// GetFuncSvcFromPoolCache has not been implemented for Container Functions
-func (caaf *Container) GetFuncSvcFromPoolCache(ctx context.Context, fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) {
-	// Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added.
-	return nil, 0, nil
-}
-
 // TapService makes a TouchByAddress request to the cache.
 func (caaf *Container) TapService(ctx context.Context, svcHost string) error {
 	err := caaf.fsCache.TouchByAddress(svcHost)

pkg/executor/executortype/executortype.go

@@ -38,9 +38,6 @@ type ExecutorType interface {
 	// GetFuncSvcFromCache retrieves function service from cache.
 	GetFuncSvcFromCache(context.Context, *fv1.Function) (*fscache.FuncSvc, error)
 
-	// GetFuncSvcFromPoolCache retrieves function service and number of active instances after filtering on requestsPerPod and CPULimit
-	GetFuncSvcFromPoolCache(ctx context.Context, fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error)
-
 	// DeleteFuncSvcFromCache deletes function service entry in cache.
 	DeleteFuncSvcFromCache(context.Context, *fscache.FuncSvc)
 

pkg/executor/executortype/newdeploy/newdeploy.go

@@ -136,7 +136,7 @@ func (deploy *NewDeploy) getDeploymentSpec(ctx context.Context, fn *fv1.Function
 	}
 
 	gracePeriodSeconds := int64(6 * 60)
-	if env.Spec.TerminationGracePeriod > 0 {
+	if env.Spec.TerminationGracePeriod >= 0 {
 		gracePeriodSeconds = env.Spec.TerminationGracePeriod
 	}
 

pkg/executor/executortype/newdeploy/newdeploymgr.go

@@ -199,12 +199,6 @@ func (deploy *NewDeploy) UnTapService(ctx context.Context, key string, svcHost s
 	// Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added.
 }
 
-// GetFuncSvcFromPoolCache has not been implemented for NewDeployment
-func (deploy *NewDeploy) GetFuncSvcFromPoolCache(ctx context.Context, fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) {
-	// Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added.
-	return nil, 0, nil
-}
-
 // TapService makes a TouchByAddress request to the cache.
 func (deploy *NewDeploy) TapService(ctx context.Context, svcHost string) error {
 	otelUtils.SpanTrackEvent(ctx, "TapService")

pkg/executor/executortype/poolmgr/gp.go

@@ -32,6 +32,7 @@ import (
 	"go.uber.org/zap"
 	appsv1 "k8s.io/api/apps/v1"
 	apiv1 "k8s.io/api/core/v1"
+	k8s_err "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -61,7 +62,7 @@ type (
 		logger                   *zap.Logger
 		env                      *fv1.Environment
 		deployment               *appsv1.Deployment            // kubernetes deployment
-		namespace                string                        // namespace to keep our resources
+		fnNamespace              string                        // namespace to keep our resources
 		podReadyTimeout          time.Duration                 // timeout for generic pods to become ready
 		fsCache                  *fscache.FunctionServiceCache // cache funcSvc's by function, address and podname
 		useSvc                   bool                          // create k8s service for specialized pods
@@ -90,7 +91,7 @@ func MakeGenericPool(
 	kubernetesClient kubernetes.Interface,
 	metricsClient metricsclient.Interface,
 	env *fv1.Environment,
-	namespace string,
+	fnNamespace string,
 	fsCache *fscache.FunctionServiceCache,
 	fetcherConfig *fetcherConfig.Config,
 	instanceID string,
@@ -119,7 +120,7 @@ func MakeGenericPool(
 		fissionClient:            fissionClient,
 		kubernetesClient:         kubernetesClient,
 		metricsClient:            metricsClient,
-		namespace:                namespace,
+		fnNamespace:              fnNamespace,
 		podReadyTimeout:          podReadyTimeout,
 		fsCache:                  fsCache,
 		fetcherConfig:            fetcherConfig,
@@ -186,7 +187,7 @@ func (gp *GenericPool) updateCPUUtilizationSvc(ctx context.Context) {
 	}
 
 	serviceFunc := func(ctx context.Context) {
-		podMetricsList, err := gp.metricsClient.MetricsV1beta1().PodMetricses(gp.namespace).List(ctx, metav1.ListOptions{
+		podMetricsList, err := gp.metricsClient.MetricsV1beta1().PodMetricses(gp.fnNamespace).List(ctx, metav1.ListOptions{
 			LabelSelector: "managed=false",
 		})
 		if err != nil {
@@ -291,7 +292,12 @@ func (gp *GenericPool) choosePod(ctx context.Context, newLabels map[string]strin
 			patch := fmt.Sprintf(`{"metadata":{"annotations":%v, "labels":%v}}`, string(annotationPatch), string(labelPatch))
 			logger.Info("relabel pod", zap.String("pod", patch))
 			newPod, err := gp.kubernetesClient.CoreV1().Pods(chosenPod.Namespace).Patch(ctx, chosenPod.Name, k8sTypes.StrategicMergePatchType, []byte(patch), metav1.PatchOptions{})
-			if err != nil {
+			if err != nil && errors.Is(err, context.Canceled) {
+				// ending retry loop when the request canceled
+				gp.readyPodQueue.Done(key)
+				gp.readyPodQueue.AddAfter(key, expoDelay)
+				return "", nil, errors.Errorf("failed to relabel pod: %s", err)
+			} else if err != nil {
 				logger.Error("failed to relabel pod", zap.Error(err), zap.String("pod", chosenPod.Name), zap.Duration("delay", expoDelay))
 				gp.readyPodQueue.Done(key)
 				gp.readyPodQueue.AddAfter(key, expoDelay)
@@ -338,12 +344,12 @@ func (gp *GenericPool) scheduleDeletePod(ctx context.Context, name string) {
 	// cleaned up.  (We need a better solutions for both those things; log
 	// aggregation and storage will help.)
 	gp.logger.Error("error in pod - scheduling cleanup", zap.String("pod", name))
-	err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(ctx, name, metav1.DeleteOptions{})
+	err := gp.kubernetesClient.CoreV1().Pods(gp.fnNamespace).Delete(ctx, name, metav1.DeleteOptions{})
 	if err != nil {
 		gp.logger.Error(
 			"error deleting pod",
 			zap.String("name", name),
-			zap.String("namespace", gp.namespace),
+			zap.String("namespace", gp.fnNamespace),
 			zap.Error(err),
 		)
 	}
@@ -387,10 +393,41 @@ func (gp *GenericPool) specializePod(ctx context.Context, pod *apiv1.Pod, fn *fv
 	if len(podIP) == 0 {
 		return errors.Errorf("Pod %s in namespace %s has no IP", pod.ObjectMeta.Name, pod.ObjectMeta.Namespace)
 	}
+	for _, cm := range fn.Spec.ConfigMaps {
+		_, err := gp.kubernetesClient.CoreV1().ConfigMaps(gp.fnNamespace).Get(ctx, cm.Name, metav1.GetOptions{})
+		if err != nil {
+			if k8s_err.IsNotFound(err) {
+				logger.Error("configmap namespace mismatch", zap.String("error", "configmap must be in same namespace as function namespace"),
+					zap.String("configmap_name", cm.Name),
+					zap.String("configmap_namespace", cm.Namespace),
+					zap.String("function_name", fn.ObjectMeta.Name),
+					zap.String("function_namespace", gp.fnNamespace))
+				return fmt.Errorf(fmt.Sprintf("configmap %s must be in same namespace as function namespace", cm.Name))
+			} else {
+				return err
+			}
+		}
+	}
+	for _, sec := range fn.Spec.Secrets {
+		_, err := gp.kubernetesClient.CoreV1().Secrets(gp.fnNamespace).Get(ctx, sec.Name, metav1.GetOptions{})
+		if err != nil {
+			if k8s_err.IsNotFound(err) {
+				logger.Error("secret namespace mismatch", zap.String("error", "secret must be in same namespace as function namespace"),
+					zap.String("secret_name", sec.Name),
+					zap.String("secret_namespace", sec.Namespace),
+					zap.String("function_name", fn.ObjectMeta.Name),
+					zap.String("function_namespace", gp.fnNamespace))
+				return fmt.Errorf(fmt.Sprintf("secret %s must be in same namespace as function namespace", sec.Name))
+			} else {
+				return err
+			}
+
+		}
+	}
 	// specialize pod with service
 	if gp.useIstio {
 		svc := utils.GetFunctionIstioServiceName(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace)
-		podIP = fmt.Sprintf("%v.%v", svc, gp.namespace)
+		podIP = fmt.Sprintf("%v.%v", svc, gp.fnNamespace)
 	}
 
 	// tell fetcher to get the function.
@@ -432,7 +469,7 @@ func (gp *GenericPool) createSvc(ctx context.Context, name string, labels map[st
 			Selector: labels,
 		},
 	}
-	svc, err := gp.kubernetesClient.CoreV1().Services(gp.namespace).Create(ctx, &service, metav1.CreateOptions{})
+	svc, err := gp.kubernetesClient.CoreV1().Services(gp.fnNamespace).Create(ctx, &service, metav1.CreateOptions{})
 	return svc, err
 }
 
@@ -470,7 +507,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac
 			"functionName": fn.ObjectMeta.Name,
 			"functionUid":  string(fn.ObjectMeta.UID),
 		}
-		podList, err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).List(ctx, metav1.ListOptions{
+		podList, err := gp.kubernetesClient.CoreV1().Pods(gp.fnNamespace).List(ctx, metav1.ListOptions{
 			LabelSelector: labels.Set(sel).AsSelector().String(),
 		})
 		if err != nil {
@@ -480,7 +517,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac
 		// Remove old versions function pods
 		for _, pod := range podList.Items {
 			// Delete pod no matter what status it is
-			gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(ctx, pod.ObjectMeta.Name, metav1.DeleteOptions{}) //nolint errcheck
+			gp.kubernetesClient.CoreV1().Pods(gp.fnNamespace).Delete(ctx, pod.ObjectMeta.Name, metav1.DeleteOptions{}) //nolint errcheck
 		}
 	}
 
@@ -515,10 +552,10 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac
 
 		// the fission router isn't in the same namespace, so return a
 		// namespace-qualified hostname
-		svcHost = fmt.Sprintf("%v.%v:8888", svcName, gp.namespace)
+		svcHost = fmt.Sprintf("%v.%v:8888", svcName, gp.fnNamespace)
 	} else if gp.useIstio {
 		svc := utils.GetFunctionIstioServiceName(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace)
-		svcHost = fmt.Sprintf("%v.%v:8888", svc, gp.namespace)
+		svcHost = fmt.Sprintf("%v.%v:8888", svc, gp.fnNamespace)
 	} else {
 		svcHost = fmt.Sprintf("%v:8888", pod.Status.PodIP)
 	}
@@ -575,8 +612,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac
 
 	gp.fsCache.PodToFsvc.Store(pod.GetObjectMeta().GetName(), fsvc)
 	gp.podFSVCMap.Store(pod.ObjectMeta.Name, []interface{}{crd.CacheKey(fsvc.Function), fsvc.Address})
-	gp.fsCache.AddFunc(ctx, *fsvc)
+	gp.fsCache.AddFunc(ctx, *fsvc, fn.GetRequestPerPod())
-
 	metrics.ColdStarts.WithLabelValues(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace).Inc()
 
 	logger.Info("added function service",
@@ -605,12 +641,12 @@ func (gp *GenericPool) destroy(ctx context.Context) error {
 	}
 
 	err := gp.kubernetesClient.AppsV1().
-		Deployments(gp.namespace).Delete(ctx, gp.deployment.ObjectMeta.Name, delOpt)
+		Deployments(gp.fnNamespace).Delete(ctx, gp.deployment.ObjectMeta.Name, delOpt)
 	if err != nil {
 		gp.logger.Error("error destroying deployment",
 			zap.Error(err),
 			zap.String("deployment_name", gp.deployment.ObjectMeta.Name),
-			zap.String("deployment_namespace", gp.namespace))
+			zap.String("deployment_namespace", gp.fnNamespace))
 		return err
 	}
 	return nil

pkg/executor/executortype/poolmgr/gp_deployment.go

@@ -70,7 +70,7 @@ func (gp *GenericPool) genDeploymentSpec(env *fv1.Environment) (*appsv1.Deployme
 	// Use long terminationGracePeriodSeconds for connection draining in case that
 	// pod still runs user functions.
 	gracePeriodSeconds := int64(6 * 60)
-	if env.Spec.TerminationGracePeriod > 0 {
+	if env.Spec.TerminationGracePeriod >= 0 {
 		gracePeriodSeconds = env.Spec.TerminationGracePeriod
 	}
 
@@ -205,13 +205,13 @@ func (gp *GenericPool) createPoolDeployment(ctx context.Context, env *fv1.Enviro
 		ObjectMeta: deploymentMeta,
 		Spec:       *deploymentSpec,
 	}
-	depl, err := gp.kubernetesClient.AppsV1().Deployments(gp.namespace).Get(ctx, deployment.Name, metav1.GetOptions{})
+	depl, err := gp.kubernetesClient.AppsV1().Deployments(gp.fnNamespace).Get(ctx, deployment.Name, metav1.GetOptions{})
 	if err == nil {
 		if depl.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != gp.instanceID {
 			deployment.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] = gp.instanceID
 			// Update with the latest deployment spec. Kubernetes will trigger
 			// rolling update if spec is different from the one in the cluster.
-			depl, err = gp.kubernetesClient.AppsV1().Deployments(gp.namespace).Update(ctx, deployment, metav1.UpdateOptions{})
+			depl, err = gp.kubernetesClient.AppsV1().Deployments(gp.fnNamespace).Update(ctx, deployment, metav1.UpdateOptions{})
 		}
 		gp.deployment = depl
 		return err
@@ -220,7 +220,7 @@ func (gp *GenericPool) createPoolDeployment(ctx context.Context, env *fv1.Enviro
 		return err
 	}
 
-	depl, err = gp.kubernetesClient.AppsV1().Deployments(gp.namespace).Create(ctx, deployment, metav1.CreateOptions{})
+	depl, err = gp.kubernetesClient.AppsV1().Deployments(gp.fnNamespace).Create(ctx, deployment, metav1.CreateOptions{})
 	if err != nil {
 		gp.logger.Error("error creating deployment in kubernetes", zap.Error(err), zap.String("deployment", deployment.Name))
 		return err
@@ -256,7 +256,7 @@ func (gp *GenericPool) updatePoolDeployment(ctx context.Context, env *fv1.Enviro
 	}
 	newDeployment.Spec.Replicas = &poolsize
 
-	depl, err := gp.kubernetesClient.AppsV1().Deployments(gp.namespace).Update(ctx, newDeployment, metav1.UpdateOptions{})
+	depl, err := gp.kubernetesClient.AppsV1().Deployments(gp.fnNamespace).Update(ctx, newDeployment, metav1.UpdateOptions{})
 	if err != nil {
 		logger.Error("error updating deployment in kubernetes", zap.Error(err), zap.String("deployment", depl.Name))
 		return err

pkg/executor/executortype/poolmgr/gpm.go

@@ -189,6 +189,7 @@ func (gpm *GenericPoolManager) GetTypeName(ctx context.Context) fv1.ExecutorType
 func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) {
 	otelUtils.SpanTrackEvent(ctx, "GetFuncSvc", otelUtils.GetAttributesForFunction(fn)...)
 	logger := otelUtils.LoggerWithTraceID(ctx, gpm.logger)
+
 	// from Func -> get Env
 	logger.Debug("getting environment for function", zap.String("function", fn.ObjectMeta.Name))
 	env, err := gpm.getFunctionEnv(ctx, fn)
@@ -212,12 +213,8 @@ func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, fn *fv1.Function)
 }
 
 func (gpm *GenericPoolManager) GetFuncSvcFromCache(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) {
-	return nil, nil
+	otelUtils.SpanTrackEvent(ctx, "GetFuncSvcFromCache", otelUtils.GetAttributesForFunction(fn)...)
-}
+	return gpm.fsCache.GetFuncSvc(ctx, &fn.ObjectMeta, fn.GetRequestPerPod(), fn.GetConcurrency())
-
-func (gpm *GenericPoolManager) GetFuncSvcFromPoolCache(ctx context.Context, fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) {
-	otelUtils.SpanTrackEvent(ctx, "GetFuncSvcFromPoolCache", otelUtils.GetAttributesForFunction(fn)...)
-	return gpm.fsCache.GetFuncSvc(ctx, &fn.ObjectMeta, requestsPerPod)
 }
 
 func (gpm *GenericPoolManager) DeleteFuncSvcFromCache(ctx context.Context, fsvc *fscache.FuncSvc) {
@@ -711,7 +708,7 @@ func (gpm *GenericPoolManager) NoActiveConnectionEventChecker(ctx context.Contex
 	defer close(stopper)
 
 	var wg wait.Group
-	for _, informer := range utils.GetInformerEventChecker(ctx, kubeClient, "WsConnectionStarted") {
+	for _, informer := range utils.GetInformerEventChecker(ctx, kubeClient, "NoActiveConnections") {
 		informer.AddEventHandler(k8sCache.ResourceEventHandlerFuncs{
 			AddFunc: func(obj interface{}) {
 				mObj := obj.(metav1.Object)

pkg/executor/executortype/poolmgr/readyPodController.go

@@ -31,7 +31,7 @@ func (gp *GenericPool) readyPodEventHandlers() k8sCache.ResourceEventHandlerFunc
 
 func (gp *GenericPool) setupReadyPodController() error {
 	gp.readyPodQueue = workqueue.NewDelayingQueue()
-	informerFactory, err := utils.GetInformerFactoryByReadyPod(gp.kubernetesClient, gp.namespace, gp.deployment.Spec.Selector)
+	informerFactory, err := utils.GetInformerFactoryByReadyPod(gp.kubernetesClient, gp.fnNamespace, gp.deployment.Spec.Selector)
 	if err != nil {
 		return err
 	}

pkg/executor/fscache/functionServiceCache.go

@@ -35,7 +35,6 @@ import (
 	"github.com/fission/fission/pkg/crd"
 	ferror "github.com/fission/fission/pkg/error"
 	"github.com/fission/fission/pkg/executor/metrics"
-	"github.com/fission/fission/pkg/poolcache"
 )
 
 type fscRequestType int
@@ -68,12 +67,12 @@ type (
 	// FunctionServiceCache represents the function service cache
 	FunctionServiceCache struct {
 		logger            *zap.Logger
-		byFunction        *cache.Cache     // function-key -> funcSvc  : map[string]*funcSvc
+		byFunction        *cache.Cache // function-key -> funcSvc  : map[string]*funcSvc
-		byAddress         *cache.Cache     // address      -> function : map[string]metav1.ObjectMeta
+		byAddress         *cache.Cache // address      -> function : map[string]metav1.ObjectMeta
-		byFunctionUID     *cache.Cache     // function uid -> function : map[string]metav1.ObjectMeta
+		byFunctionUID     *cache.Cache // function uid -> function : map[string]metav1.ObjectMeta
-		connFunctionCache *poolcache.Cache // function-key -> funcSvc : map[string]*funcSvc
+		connFunctionCache *PoolCache   // function-key -> funcSvc : map[string]*funcSvc
-		PodToFsvc         sync.Map         // pod-name -> funcSvc: map[string]*FuncSvc
+		PodToFsvc         sync.Map     // pod-name -> funcSvc: map[string]*FuncSvc
-		WebsocketFsvc     sync.Map         // funcSvc-name -> bool: map[string]bool
+		WebsocketFsvc     sync.Map     // funcSvc-name -> bool: map[string]bool
 		requestChannel    chan *fscRequest
 	}
 
@@ -113,7 +112,7 @@ func MakeFunctionServiceCache(logger *zap.Logger) *FunctionServiceCache {
 		byFunction:        cache.MakeCache(0, 0),
 		byAddress:         cache.MakeCache(0, 0),
 		byFunctionUID:     cache.MakeCache(0, 0),
-		connFunctionCache: poolcache.NewPoolCache(logger.Named("conn_function_cache")),
+		connFunctionCache: NewPoolCache(logger.Named("conn_function_cache")),
 		requestChannel:    make(chan *fscRequest),
 	}
 	go fsc.service()
@@ -159,8 +158,8 @@ func (fsc *FunctionServiceCache) service() {
 		case LISTOLDPOOL:
 			fscs := fsc.connFunctionCache.ListAvailableValue()
 			funcObjects := make([]*FuncSvc, 0)
-			for _, funcSvc := range fscs {
+			for _, fsvc := range fscs {
-				if fsvc, ok := funcSvc.(*FuncSvc); ok && time.Since(fsvc.Atime) > req.age {
+				if time.Since(fsvc.Atime) > req.age {
 					funcObjects = append(funcObjects, fsvc)
 				}
 			}
@@ -189,21 +188,20 @@ func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc,
 }
 
 // GetFuncSvc gets a function service from pool cache using function key and returns number of active instances of function pod
-func (fsc *FunctionServiceCache) GetFuncSvc(ctx context.Context, m *metav1.ObjectMeta, requestsPerPod int) (*FuncSvc, int, error) {
+func (fsc *FunctionServiceCache) GetFuncSvc(ctx context.Context, m *metav1.ObjectMeta, requestsPerPod int, concurrency int) (*FuncSvc, error) {
 	key := crd.CacheKey(m)
 
-	fsvcI, active, err := fsc.connFunctionCache.GetValue(ctx, key, requestsPerPod)
+	fsvc, err := fsc.connFunctionCache.GetSvcValue(ctx, key, requestsPerPod, concurrency)
 	if err != nil {
 		fsc.logger.Info("Not found in Cache")
-		return nil, active, err
+		return nil, err
 	}
 
 	// update atime
-	fsvc := fsvcI.(*FuncSvc)
 	fsvc.Atime = time.Now()
 
 	fsvcCopy := *fsvc
-	return &fsvcCopy, active, nil
+	return &fsvcCopy, nil
 }
 
 // GetByFunctionUID gets a function service from cache using function UUID.
@@ -229,8 +227,8 @@ func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, erro
 }
 
 // AddFunc adds a function service to pool cache.
-func (fsc *FunctionServiceCache) AddFunc(ctx context.Context, fsvc FuncSvc) {
+func (fsc *FunctionServiceCache) AddFunc(ctx context.Context, fsvc FuncSvc, requestsPerPod int) {
-	fsc.connFunctionCache.SetValue(ctx, crd.CacheKey(fsvc.Function), fsvc.Address, &fsvc, fsvc.CPULimit)
+	fsc.connFunctionCache.SetSvcValue(ctx, crd.CacheKey(fsvc.Function), fsvc.Address, &fsvc, fsvc.CPULimit, requestsPerPod)
 	now := time.Now()
 	fsvc.Ctime = now
 	fsvc.Atime = now

pkg/executor/fscache/functionServiceCache_test.go

@@ -187,19 +187,17 @@ func TestFunctionServiceNewCache(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	fsc.AddFunc(ctx, *fsvc)
+	fsc.AddFunc(ctx, *fsvc, 10)
-	_, active, err := fsc.GetFuncSvc(ctx, fsvc.Function, 5)
+	concurrency := 10
+	_, err = fsc.GetFuncSvc(ctx, fsvc.Function, 5, concurrency)
 	if err != nil {
 		logger.Panic("received error while retrieving value from cache")
 	}
-	if active != 1 {
-		logger.Panic(fmt.Sprintln("active instances not matched expected 1, found ", active))
-	}
 
 	key := fmt.Sprintf("%v_%v", fn.ObjectMeta.UID, fn.ObjectMeta.ResourceVersion)
 	fsc.MarkAvailable(key, fsvc.Address)
 
-	_, _, err = fsc.GetFuncSvc(ctx, fsvc.Function, 5)
+	_, err = fsc.GetFuncSvc(ctx, fsvc.Function, 5, concurrency)
 	if err != nil {
 		logger.Panic("received error while retrieving value from cache")
 	}

pkg/executor/fscache/poolcache.go

@@ -0,0 +1,330 @@
+/*
+Copyright 2016 The Fission Authors.
+
+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 fscache
+
+import (
+	"context"
+	"fmt"
+
+	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/api/resource"
+
+	ferror "github.com/fission/fission/pkg/error"
+	otelUtils "github.com/fission/fission/pkg/utils/otel"
+)
+
+type requestType int
+
+const (
+	getValue requestType = iota
+	listAvailableValue
+	setValue
+	markAvailable
+	deleteValue
+	setCPUUtilization
+)
+
+type (
+	funcSvcInfo struct {
+		val             *FuncSvc
+		activeRequests  int               // number of requests served by function pod
+		currentCPUUsage resource.Quantity // current cpu usage of the specialized function pod
+		cpuLimit        resource.Quantity // if currentCPUUsage is more than cpuLimit cache miss occurs in getValue request
+	}
+
+	funcSvcGroup struct {
+		svcWaiting int
+		svcs       map[string]*funcSvcInfo
+		queue      *Queue
+	}
+
+	// PoolCache implements a simple cache implementation having values mapped by two keys [function][address].
+	// As of now PoolCache is only used by poolmanager executor
+	PoolCache struct {
+		cache          map[string]*funcSvcGroup
+		requestChannel chan *request
+		logger         *zap.Logger
+	}
+
+	request struct {
+		requestType
+		ctx             context.Context
+		function        string
+		address         string
+		value           *FuncSvc
+		requestsPerPod  int
+		cpuUsage        resource.Quantity
+		responseChannel chan *response
+		concurrency     int
+	}
+	response struct {
+		error
+		allValues    []*FuncSvc
+		value        *FuncSvc
+		svcWaitValue *svcWait
+	}
+	svcWait struct {
+		svcChannel chan *FuncSvc
+		ctx        context.Context
+	}
+)
+
+// NewPoolCache create a Cache object
+
+func NewPoolCache(logger *zap.Logger) *PoolCache {
+	c := &PoolCache{
+		cache:          make(map[string]*funcSvcGroup),
+		requestChannel: make(chan *request),
+		logger:         logger,
+	}
+	go c.service()
+	return c
+}
+
+func NewFuncSvcGroup() *funcSvcGroup {
+	return &funcSvcGroup{
+		svcs:  make(map[string]*funcSvcInfo),
+		queue: NewQueue(),
+	}
+}
+
+func (c *PoolCache) service() {
+	for {
+		req := <-c.requestChannel
+		resp := &response{}
+		switch req.requestType {
+		case getValue:
+			funcSvcGroup, ok := c.cache[req.function]
+			if !ok {
+				c.cache[req.function] = NewFuncSvcGroup()
+				c.cache[req.function].svcWaiting++
+				resp.error = ferror.MakeError(ferror.ErrorNotFound,
+					fmt.Sprintf("function Name '%v' not found", req.function))
+				req.responseChannel <- resp
+				continue
+			}
+			found := false
+			totalActiveRequests := 0
+			for addr := range funcSvcGroup.svcs {
+				totalActiveRequests += funcSvcGroup.svcs[addr].activeRequests
+				if funcSvcGroup.svcs[addr].activeRequests < req.requestsPerPod &&
+					funcSvcGroup.svcs[addr].currentCPUUsage.Cmp(funcSvcGroup.svcs[addr].cpuLimit) < 1 {
+					// mark active
+					funcSvcGroup.svcs[addr].activeRequests++
+					if c.logger.Core().Enabled(zap.DebugLevel) {
+						otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Increase active requests with getValue", zap.String("function", req.function), zap.String("address", addr), zap.Int("activeRequests", funcSvcGroup.svcs[addr].activeRequests))
+					}
+					resp.value = funcSvcGroup.svcs[addr].val
+					found = true
+					break
+				}
+			}
+			if found {
+				req.responseChannel <- resp
+				continue
+			}
+			specializationInProgress := funcSvcGroup.svcWaiting - funcSvcGroup.queue.Len()
+			capacity := ((specializationInProgress + len(funcSvcGroup.svcs)) * req.requestsPerPod) - (totalActiveRequests + funcSvcGroup.svcWaiting)
+			if capacity > 0 {
+				funcSvcGroup.svcWaiting++
+				svcWait := &svcWait{
+					svcChannel: make(chan *FuncSvc),
+					ctx:        req.ctx,
+				}
+				resp.svcWaitValue = svcWait
+				funcSvcGroup.queue.Push(svcWait)
+				req.responseChannel <- resp
+				continue
+			}
+
+			// concurrency should not be set to zero and
+			//sum of specialization in progress and specialized pods should be less then req.concurrency
+			if req.concurrency > 0 && (specializationInProgress+len(funcSvcGroup.svcs)) >= req.concurrency {
+				resp.error = ferror.MakeError(ferror.ErrorTooManyRequests, fmt.Sprintf("function '%s' concurrency '%d' limit reached.", req.function, req.concurrency))
+			} else {
+				funcSvcGroup.svcWaiting++
+				resp.error = ferror.MakeError(ferror.ErrorNotFound, fmt.Sprintf("function '%s' all functions are busy", req.function))
+			}
+			req.responseChannel <- resp
+		case setValue:
+			if _, ok := c.cache[req.function]; !ok {
+				c.cache[req.function] = NewFuncSvcGroup()
+			}
+			if _, ok := c.cache[req.function].svcs[req.address]; !ok {
+				c.cache[req.function].svcs[req.address] = &funcSvcInfo{}
+			}
+			c.cache[req.function].svcs[req.address].val = req.value
+			c.cache[req.function].svcs[req.address].activeRequests++
+			if c.cache[req.function].svcWaiting > 0 {
+				c.cache[req.function].svcWaiting--
+				svcCapacity := req.requestsPerPod - c.cache[req.function].svcs[req.address].activeRequests
+				queueLen := c.cache[req.function].queue.Len()
+				if svcCapacity > queueLen {
+					svcCapacity = queueLen
+				}
+				for i := 0; i <= svcCapacity; {
+					popped := c.cache[req.function].queue.Pop()
+					if popped == nil {
+						break
+					}
+					if popped.ctx.Err() == nil {
+						popped.svcChannel <- req.value
+						c.cache[req.function].svcs[req.address].activeRequests++
+						i++
+					}
+					close(popped.svcChannel)
+					c.cache[req.function].svcWaiting--
+				}
+			}
+			if c.logger.Core().Enabled(zap.DebugLevel) {
+				otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Increase active requests with setValue", zap.String("function", req.function), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function].svcs[req.address].activeRequests))
+			}
+			c.cache[req.function].svcs[req.address].cpuLimit = req.cpuUsage
+		case listAvailableValue:
+			vals := make([]*FuncSvc, 0)
+			for key1, values := range c.cache {
+				for key2, value := range values.svcs {
+					debugLevel := c.logger.Core().Enabled(zap.DebugLevel)
+					if debugLevel {
+						otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Reading active requests", zap.String("function", key1), zap.String("address", key2), zap.Int("activeRequests", value.activeRequests))
+					}
+					if value.activeRequests == 0 {
+						if debugLevel {
+							otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Function service with no active requests", zap.String("function", key1), zap.String("address", key2), zap.Int("activeRequests", value.activeRequests))
+						}
+						vals = append(vals, value.val)
+					}
+				}
+			}
+			resp.allValues = vals
+			req.responseChannel <- resp
+		case setCPUUtilization:
+			if _, ok := c.cache[req.function]; !ok {
+				c.cache[req.function] = NewFuncSvcGroup()
+			}
+			if _, ok := c.cache[req.function].svcs[req.address]; ok {
+				c.cache[req.function].svcs[req.address].currentCPUUsage = req.cpuUsage
+			}
+		case markAvailable:
+			if _, ok := c.cache[req.function]; ok {
+				if _, ok = c.cache[req.function].svcs[req.address]; ok {
+					if c.cache[req.function].svcs[req.address].activeRequests > 0 {
+						c.cache[req.function].svcs[req.address].activeRequests--
+						if c.logger.Core().Enabled(zap.DebugLevel) {
+							otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Decrease active requests", zap.String("function", req.function), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function].svcs[req.address].activeRequests))
+						}
+					} else {
+						otelUtils.LoggerWithTraceID(req.ctx, c.logger).Error("Invalid request to decrease active requests", zap.String("function", req.function), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function].svcs[req.address].activeRequests))
+					}
+				}
+			}
+		case deleteValue:
+			delete(c.cache[req.function].svcs, req.address)
+			req.responseChannel <- resp
+		default:
+			resp.error = ferror.MakeError(ferror.ErrorInvalidArgument,
+				fmt.Sprintf("invalid request type: %v", req.requestType))
+			req.responseChannel <- resp
+		}
+	}
+}
+
+// GetValue returns a function service with status in Active else return error
+func (c *PoolCache) GetSvcValue(ctx context.Context, function string, requestsPerPod int, concurrency int) (*FuncSvc, error) {
+	respChannel := make(chan *response)
+	c.requestChannel <- &request{
+		ctx:             ctx,
+		requestType:     getValue,
+		function:        function,
+		concurrency:     concurrency,
+		requestsPerPod:  requestsPerPod,
+		responseChannel: respChannel,
+	}
+	resp := <-respChannel
+
+	if resp.svcWaitValue != nil {
+		select {
+		case <-ctx.Done():
+			return resp.value, ctx.Err()
+		case funcSvc := <-resp.svcWaitValue.svcChannel:
+			return funcSvc, nil
+		}
+	}
+	return resp.value, resp.error
+}
+
+// ListAvailableValue returns a list of the available function services stored in the Cache
+func (c *PoolCache) ListAvailableValue() []*FuncSvc {
+	respChannel := make(chan *response)
+	c.requestChannel <- &request{
+		requestType:     listAvailableValue,
+		responseChannel: respChannel,
+	}
+	resp := <-respChannel
+	return resp.allValues
+}
+
+// SetValue marks the value at key [function][address] as active(begin used)
+func (c *PoolCache) SetSvcValue(ctx context.Context, function, address string, value *FuncSvc, cpuLimit resource.Quantity, requestsPerPod int) {
+	respChannel := make(chan *response)
+	c.requestChannel <- &request{
+		ctx:             ctx,
+		requestType:     setValue,
+		function:        function,
+		address:         address,
+		value:           value,
+		cpuUsage:        cpuLimit,
+		requestsPerPod:  requestsPerPod,
+		responseChannel: respChannel,
+	}
+}
+
+// SetCPUUtilization updates/sets the CPU utilization limit for the pod
+func (c *PoolCache) SetCPUUtilization(function, address string, cpuUsage resource.Quantity) {
+	c.requestChannel <- &request{
+		requestType:     setCPUUtilization,
+		function:        function,
+		address:         address,
+		cpuUsage:        cpuUsage,
+		responseChannel: make(chan *response),
+	}
+}
+
+// MarkAvailable marks the value at key [function][address] as available
+func (c *PoolCache) MarkAvailable(function, address string) {
+	respChannel := make(chan *response)
+	c.requestChannel <- &request{
+		requestType:     markAvailable,
+		function:        function,
+		address:         address,
+		responseChannel: respChannel,
+	}
+}
+
+// DeleteValue deletes the value at key composed of [function][address]
+func (c *PoolCache) DeleteValue(ctx context.Context, function, address string) error {
+	respChannel := make(chan *response)
+	c.requestChannel <- &request{
+		ctx:             ctx,
+		requestType:     deleteValue,
+		function:        function,
+		address:         address,
+		responseChannel: respChannel,
+	}
+	resp := <-respChannel
+	return resp.error
+}

pkg/executor/fscache/poolcache_test.go

@@ -0,0 +1,200 @@
+package fscache
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"net/http"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/api/resource"
+
+	ferror "github.com/fission/fission/pkg/error"
+	"github.com/fission/fission/pkg/utils/loggerfactory"
+)
+
+func checkErr(err error) {
+	if err != nil {
+		log.Panicf("err: %v", err)
+	}
+}
+
+func TestPoolCache(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	logger := loggerfactory.GetLogger()
+	c := NewPoolCache(logger)
+	concurrency := 5
+	requestsPerPod := 2
+
+	// should return err since no svc is present
+	_, err := c.GetSvcValue(ctx, "func", requestsPerPod, concurrency)
+	if err == nil {
+		log.Panicf("found value when expected it to be nil")
+	}
+
+	c.SetSvcValue(ctx, "func", "ip", &FuncSvc{
+		Name: "value",
+	}, resource.MustParse("45m"), 10)
+
+	// should not return any error since we added a svc
+	_, err = c.GetSvcValue(ctx, "func", requestsPerPod, concurrency)
+	checkErr(err)
+
+	c.SetSvcValue(ctx, "func", "ip", &FuncSvc{
+		Name: "value",
+	}, resource.MustParse("45m"), 10)
+
+	// should return err since all functions are busy
+	_, err = c.GetSvcValue(ctx, "func", requestsPerPod, concurrency)
+	if err == nil {
+		log.Panicf("found value when expected it to be nil")
+	}
+
+	c.SetSvcValue(ctx, "func", "ip", &FuncSvc{
+		Name: "value",
+	}, resource.MustParse("45m"), 10)
+
+	c.SetSvcValue(ctx, "func2", "ip2", &FuncSvc{
+		Name: "value2",
+	}, resource.MustParse("50m"), 10)
+
+	c.SetSvcValue(ctx, "func2", "ip22", &FuncSvc{
+		Name: "value22",
+	}, resource.MustParse("33m"), 10)
+
+	checkErr(c.DeleteValue(ctx, "func2", "ip2"))
+
+	cc := c.ListAvailableValue()
+	if len(cc) != 0 {
+		log.Panicf("expected 0 available items")
+	}
+
+	c.MarkAvailable("func", "ip")
+
+	checkErr(c.DeleteValue(ctx, "func", "ip"))
+
+	_, err = c.GetSvcValue(ctx, "func", requestsPerPod, concurrency)
+	if err == nil {
+		log.Panicf("found deleted element")
+	}
+
+	c.SetSvcValue(ctx, "cpulimit", "100", &FuncSvc{
+		Name: "value",
+	}, resource.MustParse("3m"), 10)
+	c.SetCPUUtilization("cpulimit", "100", resource.MustParse("4m"))
+}
+
+func TestPoolCacheRequests(t *testing.T) {
+
+	type structForTest struct {
+		name           string
+		requests       int
+		concurrency    int
+		rpp            int
+		simultaneous   int
+		failedRequests int
+	}
+
+	for _, tt := range []structForTest{
+		{
+			name:        "test1",
+			requests:    1,
+			concurrency: 1,
+			rpp:         1,
+		},
+		{
+			name:        "test2",
+			requests:    2,
+			concurrency: 2,
+			rpp:         1,
+		},
+		{
+			name:        "test3",
+			requests:    300,
+			concurrency: 5,
+			rpp:         60,
+		},
+
+		{
+			name:           "test4",
+			requests:       6,
+			concurrency:    1,
+			rpp:            5,
+			failedRequests: 1,
+		},
+		{
+			name:           "test5",
+			requests:       6,
+			concurrency:    5,
+			rpp:            1,
+			failedRequests: 1,
+		},
+		{
+			name:         "test6",
+			requests:     300,
+			concurrency:  5,
+			rpp:          60,
+			simultaneous: 30,
+		},
+		{
+			name:           "test7",
+			requests:       310,
+			concurrency:    5,
+			rpp:            60,
+			simultaneous:   30,
+			failedRequests: 10,
+		},
+		{
+			name:         "test8",
+			requests:     10,
+			concurrency:  10,
+			rpp:          1,
+			simultaneous: 10,
+		},
+	} {
+		t.Run(fmt.Sprintf("scenario-%s", tt.name), func(t *testing.T) {
+			var failedRequests, svcCounter uint64
+			p := NewPoolCache(loggerfactory.GetLogger())
+			wg := sync.WaitGroup{}
+			simultaneous := tt.simultaneous
+			if simultaneous == 0 {
+				simultaneous = 1
+			}
+			for i := 1; i <= tt.requests; i++ {
+				wg.Add(1)
+				go func(reqno int) {
+					defer wg.Done()
+					svc, err := p.GetSvcValue(context.Background(), "func", tt.rpp, tt.concurrency)
+					if err != nil {
+						code, _ := ferror.GetHTTPError(err)
+						if code == http.StatusNotFound {
+							p.SetSvcValue(context.Background(), "func", fmt.Sprintf("svc-%d", svcCounter), &FuncSvc{
+								Name: "value",
+							}, resource.MustParse("45m"), tt.rpp)
+							atomic.AddUint64(&svcCounter, 1)
+						} else {
+							t.Log(reqno, "=>", err)
+							atomic.AddUint64(&failedRequests, 1)
+						}
+					} else {
+						if svc == nil {
+							t.Log(reqno, "=>", "svc is nil")
+							atomic.AddUint64(&failedRequests, 1)
+						}
+					}
+				}(i)
+				if i%simultaneous == 0 {
+					wg.Wait()
+				}
+			}
+			wg.Wait()
+
+			require.Equal(t, tt.failedRequests, int(atomic.LoadUint64(&failedRequests)))
+			require.Equal(t, tt.concurrency, int(atomic.LoadUint64(&svcCounter)))
+		})
+	}
+}

pkg/executor/fscache/queue.go

@@ -0,0 +1,45 @@
+package fscache
+
+import (
+	"container/list"
+	"sync"
+)
+
+type Queue struct {
+	items *list.List
+	mutex sync.Mutex
+}
+
+func NewQueue() *Queue {
+	return &Queue{
+		items: list.New(),
+	}
+}
+
+func (q *Queue) Push(item *svcWait) {
+	q.mutex.Lock()
+	defer q.mutex.Unlock()
+	q.items.PushBack(item)
+}
+
+func (q *Queue) Pop() *svcWait {
+	q.mutex.Lock()
+	defer q.mutex.Unlock()
+
+	item := q.items.Front()
+	if item == nil {
+		return nil
+	}
+	q.items.Remove(item)
+	svcWait, ok := item.Value.(*svcWait)
+	if !ok {
+		return nil
+	}
+	return svcWait
+}
+
+func (q *Queue) Len() int {
+	q.mutex.Lock()
+	defer q.mutex.Unlock()
+	return q.items.Len()
+}

pkg/executor/fscache/queue_test.go

@@ -0,0 +1,115 @@
+package fscache
+
+import (
+	"sync"
+	"testing"
+)
+
+func TestNewQueue(t *testing.T) {
+	q := NewQueue()
+	if q == nil {
+		t.Error("NewQueue returned nil")
+	}
+}
+
+func TestQueuePushWithSingleRequest(t *testing.T) {
+	q := NewQueue()
+	item := &svcWait{
+		svcChannel: make(chan *FuncSvc),
+		ctx:        nil,
+	}
+	q.Push(item)
+	if q.Len() != 1 {
+		t.Errorf("Expected queue length to be 1, got %d", q.Len())
+	}
+}
+
+func TestQueuePopWithSingleRequest(t *testing.T) {
+	q := NewQueue()
+	item := &svcWait{
+		svcChannel: make(chan *FuncSvc),
+		ctx:        nil,
+	}
+	q.Push(item)
+	popped := q.Pop()
+	if popped == nil {
+		t.Error("Expected Pop to return a non-nil value")
+	}
+	if popped != item {
+		t.Error("Expected Pop to return the same element that was pushed")
+	}
+	if q.Len() != 0 {
+		t.Errorf("Expected queue length to be 0, got %d", q.Len())
+	}
+}
+
+func TestQueuePushWithConcurrentRequest(t *testing.T) {
+	q := NewQueue()
+	noOfRequests := 20
+	var wg sync.WaitGroup
+	wg.Add(noOfRequests)
+	for i := 0; i < noOfRequests; i++ {
+		go func() {
+			defer wg.Done()
+			item := &svcWait{
+				svcChannel: make(chan *FuncSvc),
+				ctx:        nil,
+			}
+			q.Push(item)
+		}()
+	}
+
+	wg.Wait()
+
+	if q.Len() != noOfRequests {
+		t.Errorf("Expected queue length to be 20, got %d", q.Len())
+	}
+}
+
+func TestQueuePopWithConcurrentRequest(t *testing.T) {
+	q := NewQueue()
+	noOfPush := 20
+	noOfPop := 15
+
+	var wg sync.WaitGroup
+	wg.Add(noOfPush + noOfPop)
+
+	for i := 0; i < noOfPush; i++ {
+		go func() {
+			defer wg.Done()
+			item := &svcWait{
+				svcChannel: make(chan *FuncSvc),
+				ctx:        nil,
+			}
+			q.Push(item)
+		}()
+	}
+
+	for i := 0; i < noOfPop; i++ {
+		go func() {
+			defer wg.Done()
+			q.Pop()
+		}()
+	}
+
+	wg.Wait()
+
+	if q.Len() != 5 {
+		t.Errorf("Expected queue length to be 5, got %d", q.Len())
+	}
+}
+
+func TestQueueLen(t *testing.T) {
+	q := NewQueue()
+	if q.Len() != 0 {
+		t.Errorf("Expected queue length to be 0, got %d", q.Len())
+	}
+	item := &svcWait{
+		svcChannel: make(chan *FuncSvc),
+		ctx:        nil,
+	}
+	q.Push(item)
+	if q.Len() != 1 {
+		t.Errorf("Expected queue length to be 1, got %d", q.Len())
+	}
+}

pkg/fission-cli/cmd/function/test.go

@@ -50,14 +50,19 @@ func Test(input cli.Input) error {
 }
 
 func (opts *TestSubCommand) do(input cli.Input) error {
-
+	fnName := input.String(flagkey.FnName)
 	_, namespace, err := opts.GetResourceNamespace(input, flagkey.NamespaceFunction)
 	if err != nil {
 		return errors.Wrap(err, "error in testing function ")
 	}
 
+	function, err := opts.Client().FissionClientSet.CoreV1().Functions(namespace).Get(input.Context(), fnName, metav1.GetOptions{})
+	if err != nil {
+		return errors.Wrap(err, fmt.Sprintf("read function '%s'", fnName))
+	}
+
 	m := &metav1.ObjectMeta{
-		Name:      input.String(flagkey.FnName),
+		Name:      fnName,
 		Namespace: namespace,
 	}
 	routerURL := os.Getenv("FISSION_ROUTER")
@@ -107,14 +112,26 @@ func (opts *TestSubCommand) do(input cli.Input) error {
 		functionUrl.RawQuery = query.Encode()
 	}
 
-	var ctx context.Context
+	var (
+		ctx        context.Context
+		reqTimeout time.Duration
+	)
 
-	testTimeout := input.Duration(flagkey.FnTestTimeout)
+	fnTestTimeout := input.Duration(flagkey.FnTestTimeout)
-	if testTimeout <= 0*time.Second {
+	fnSpecTimeout := time.Duration(function.Spec.FunctionTimeout)
+
+	if input.IsSet(flagkey.FnTestTimeout) && (fnTestTimeout < fnSpecTimeout) {
+		reqTimeout = fnTestTimeout
+		console.Warn(fmt.Sprintf("timeout specified is less than functionTimeout %d Overriding value to %d", fnTestTimeout, fnSpecTimeout))
+	} else {
+		reqTimeout = fnSpecTimeout
+	}
+
+	if reqTimeout <= 0*time.Second {
 		ctx = input.Context()
 	} else {
 		var closeCtx context.CancelFunc
-		ctx, closeCtx = context.WithTimeout(input.Context(), input.Duration(flagkey.FnTestTimeout))
+		ctx, closeCtx = context.WithTimeout(input.Context(), reqTimeout*time.Second)
 		defer closeCtx()
 	}
 

pkg/fission-cli/flag/flag.go

@@ -133,7 +133,7 @@ var (
 	FnSubPath               = Flag{Type: String, Name: flagkey.FnSubPath, Usage: "Sub Path to check if function internally supports routing"}
 	FnLogAllPods            = Flag{Type: Bool, Name: flagkey.FnLogAllPods, Usage: "Get all pod's logs in the function."}
 	// Termination Grace Period configurable at function creation/update only for container functions
-	FnTerminationGracePeriod = Flag{Type: Int64, Name: flagkey.FnGracePeriod, Usage: "Grace time (in seconds) for pod to perform connection draining before termination (default value will be used if negative value is given)", DefaultValue: 360}
+	FnTerminationGracePeriod = Flag{Type: Int64, Name: flagkey.FnGracePeriod, Usage: "Grace time (in seconds) for pod to perform connection draining before termination (only non-negative values considered)", DefaultValue: 360}
 
 	HtName              = Flag{Type: String, Name: flagkey.HtName, Usage: "HTTP trigger name"}
 	HtMethod            = Flag{Type: StringSlice, Name: flagkey.HtMethod, Usage: "HTTP Methods: GET,POST,PUT,DELETE,HEAD. To mention single method: --method GET and for multiple methods --method GET --method POST. [DEPRECATED for 'fn create', use 'route create' instead]", DefaultValue: []string{http.MethodGet}}
@@ -181,7 +181,7 @@ var (
 	EnvBuildCmd               = Flag{Type: String, Name: flagkey.EnvBuildcommand, Usage: "Build command for environment builder to build source package"}
 	EnvKeepArchive            = Flag{Type: Bool, Name: flagkey.EnvKeeparchive, Usage: "Keep the archive instead of extracting it into a directory (mainly for the JVM environment because .jar is one kind of zip archive)"}
 	EnvExternalNetwork        = Flag{Type: Bool, Name: flagkey.EnvExternalNetwork, Usage: "Allow pod to access external network (only works when istio feature is enabled)"}
-	EnvTerminationGracePeriod = Flag{Type: Int64, Name: flagkey.EnvGracePeriod, Aliases: []string{"period"}, Usage: "Grace time (in seconds) for pod to perform connection draining before termination (default value will be used if 0 is given)", DefaultValue: 360}
+	EnvTerminationGracePeriod = Flag{Type: Int64, Name: flagkey.EnvGracePeriod, Aliases: []string{"period"}, Usage: "Grace time (in seconds) for pod to perform connection draining before termination (only non-negative values considered)", DefaultValue: 360}
 	EnvVersion                = Flag{Type: Int, Name: flagkey.EnvVersion, Usage: "Environment API version (1 means v1 interface)", DefaultValue: 1}
 	EnvImagePullSecret        = Flag{Type: String, Name: flagkey.EnvImagePullSecret, Usage: "Secret for Kubernetes to pull an image from a private registry"}
 	EnvExecutorType           = Flag{Type: String, Name: flagkey.EnvExecutorType, Usage: "Executor type of pod in environment; one of 'poolmgr', 'newdeploy', 'container'"}

pkg/fission-cli/util/util.go

@@ -50,8 +50,17 @@ import (
 	"github.com/fission/fission/pkg/utils"
 )
 
+const (
+	ENV_FISSION_NAMESPACE  string = "FISSION_NAMESPACE"
+	ENV_FISSION_URL        string = "FISSION_URL"
+	ENV_FISSION_AUTH_TOKEN string = "FISSION_AUTH_TOKEN"
+	localhostURL           string = "http://127.0.0.1:"
+	authHeader             string = "Authorization"
+	tokenType              string = "Bearer"
+)
+
 func GetFissionNamespace() string {
-	fissionNamespace := os.Getenv("FISSION_NAMESPACE")
+	fissionNamespace := os.Getenv(ENV_FISSION_NAMESPACE)
 	return fissionNamespace
 }
 
@@ -68,14 +77,14 @@ func ResolveFunctionNS(namespace string) string {
 func GetApplicationUrl(ctx context.Context, client cmd.Client, selector string) (string, error) {
 	var serverUrl string
 	// Use FISSION_URL env variable if set; otherwise, port-forward to controller.
-	fissionUrl := os.Getenv("FISSION_URL")
+	fissionUrl := os.Getenv(ENV_FISSION_URL)
 	if len(fissionUrl) == 0 {
 		fissionNamespace := GetFissionNamespace()
 		localPort, err := SetupPortForward(ctx, client, fissionNamespace, selector)
 		if err != nil {
 			return "", err
 		}
-		serverUrl = "http://127.0.0.1:" + localPort
+		serverUrl = fmt.Sprintf("%s%s", localhostURL, localPort)
 	} else {
 		serverUrl = fissionUrl
 	}
@@ -167,13 +176,27 @@ func GetServerInfo(input cli.Input, cmdClient cmd.Client) *info.ServerInfo {
 		return &serverInfo
 	}
 	// make request
-	resp, err := http.Get(serverURL.String() + "/_version")
+	req, err := http.NewRequestWithContext(input.Context(), "GET", fmt.Sprintf("%s%s", serverURL.String(), "/_version"), nil)
+	if err != nil {
+		console.Warn("could not create http request")
+		return &serverInfo
+	}
+
+	req.Header.Add(authHeader, fmt.Sprintf("%s %s", tokenType, os.Getenv(ENV_FISSION_AUTH_TOKEN)))
+	client := &http.Client{}
+	resp, err := client.Do(req)
 	if err != nil {
 		console.Warn("could not get data from server")
 		return &serverInfo
 	}
 
 	defer resp.Body.Close()
+	if resp.StatusCode == http.StatusForbidden || resp.StatusCode == http.StatusUnauthorized {
+		// display user a warning message to set environment variable FISSION_AUTH_TOKEN
+		if len(os.Getenv(ENV_FISSION_AUTH_TOKEN)) <= 0 {
+			console.Warn(fmt.Sprintf("Please consider setting %s as environment variable, if authentication is enabled", ENV_FISSION_AUTH_TOKEN))
+		}
+	}
 	if resp.StatusCode != http.StatusOK {
 		msg := fmt.Sprintf("HTTP error %v", resp.StatusCode)
 		console.Warn(msg)
@@ -196,7 +219,7 @@ func getRouterURL(ctx context.Context, cmdClient cmd.Client) (serverURL *url.URL
 		return serverURL, err
 	}
 
-	serverURL, err = url.Parse("http://127.0.0.1:" + localRouterPort)
+	serverURL, err = url.Parse(fmt.Sprintf("%s%s", localhostURL, localRouterPort))
 	if err != nil {
 		return serverURL, err
 	}
@@ -445,7 +468,7 @@ func GetStorageURL(ctx context.Context, client cmd.Client) (*url.URL, error) {
 		return nil, err
 	}
 
-	serverURL, err := url.Parse("http://127.0.0.1:" + storageLocalPort)
+	serverURL, err := url.Parse(fmt.Sprintf("%s%s", localhostURL, storageLocalPort))
 	if err != nil {
 		return nil, err
 	}

pkg/kubewatcher/kubewatcher.go

@@ -268,7 +268,7 @@ func (ws *watchSubscription) eventDispatchLoop(ctx context.Context) {
 		// the triggers can only be created in the same namespace as the function.
 		// so essentially, function namespace = trigger namespace.
 		url := utils.UrlForFunction(ws.watch.Spec.FunctionReference.Name, ws.watch.ObjectMeta.Namespace)
-		ws.publisher.Publish(buf.String(), headers, url)
+		ws.publisher.Publish(ctx, buf.String(), headers, url)
 	}
 }
 

pkg/poolcache/poolcache.go

@@ -1,252 +0,0 @@
-/*
-Copyright 2016 The Fission Authors.
-
-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 poolcache implements a simple cache implementation having values mapped by two keys.
-// As of now this package is only used by poolmanager executor
-package poolcache
-
-import (
-	"context"
-	"fmt"
-
-	"go.uber.org/zap"
-	"k8s.io/apimachinery/pkg/api/resource"
-
-	ferror "github.com/fission/fission/pkg/error"
-	otelUtils "github.com/fission/fission/pkg/utils/otel"
-)
-
-type requestType int
-
-const (
-	getValue requestType = iota
-	listAvailableValue
-	setValue
-	markAvailable
-	deleteValue
-	setCPUUtilization
-)
-
-type (
-	// value used as "value" in cache
-	value struct {
-		val             interface{}
-		activeRequests  int               // number of requests served by function pod
-		currentCPUUsage resource.Quantity // current cpu usage of the specialized function pod
-		cpuLimit        resource.Quantity // if currentCPUUsage is more than cpuLimit cache miss occurs in getValue request
-	}
-	// Cache is simple cache having two keys [function][address] mapped to value and requestChannel for operation on it
-	Cache struct {
-		cache          map[interface{}]map[interface{}]*value
-		requestChannel chan *request
-		logger         *zap.Logger
-	}
-
-	request struct {
-		requestType
-		ctx             context.Context
-		function        interface{}
-		address         interface{}
-		value           interface{}
-		requestsPerPod  int
-		cpuUsage        resource.Quantity
-		responseChannel chan *response
-	}
-	response struct {
-		error
-		allValues   []interface{}
-		value       interface{}
-		totalActive int
-	}
-)
-
-// NewPoolCache create a Cache object
-func NewPoolCache(logger *zap.Logger) *Cache {
-	c := &Cache{
-		cache:          make(map[interface{}]map[interface{}]*value),
-		requestChannel: make(chan *request),
-		logger:         logger,
-	}
-	go c.service()
-	return c
-}
-
-func (c *Cache) service() {
-	for {
-		req := <-c.requestChannel
-		resp := &response{}
-		switch req.requestType {
-		case getValue:
-			values, ok := c.cache[req.function]
-			found := false
-			if !ok {
-				resp.error = ferror.MakeError(ferror.ErrorNotFound,
-					fmt.Sprintf("function Name '%v' not found", req.function))
-			} else {
-				for addr := range values {
-					if values[addr].activeRequests < req.requestsPerPod && values[addr].currentCPUUsage.Cmp(values[addr].cpuLimit) < 1 {
-						// mark active
-						values[addr].activeRequests++
-						if c.logger.Core().Enabled(zap.DebugLevel) {
-							otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Increase active requests with getValue", zap.String("function", req.function.(string)), zap.String("address", addr.(string)), zap.Int("activeRequests", values[addr].activeRequests))
-						}
-						resp.value = values[addr].val
-						found = true
-						break
-					}
-				}
-				if !found {
-					resp.error = ferror.MakeError(ferror.ErrorNotFound, fmt.Sprintf("function '%v' all functions are busy", req.function))
-				}
-				resp.totalActive = len(values)
-			}
-			req.responseChannel <- resp
-		case setValue:
-			if _, ok := c.cache[req.function]; !ok {
-				c.cache[req.function] = make(map[interface{}]*value)
-			}
-			if _, ok := c.cache[req.function][req.address]; !ok {
-				c.cache[req.function][req.address] = &value{}
-			}
-			c.cache[req.function][req.address].val = req.value
-			c.cache[req.function][req.address].activeRequests++
-			if c.logger.Core().Enabled(zap.DebugLevel) {
-				otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Increase active requests with setValue", zap.String("function", req.function.(string)), zap.String("address", req.address.(string)), zap.Int("activeRequests", c.cache[req.function][req.address].activeRequests))
-			}
-			c.cache[req.function][req.address].cpuLimit = req.cpuUsage
-		case listAvailableValue:
-			vals := make([]interface{}, 0)
-			for key1, values := range c.cache {
-				for key2, value := range values {
-					debugLevel := c.logger.Core().Enabled(zap.DebugLevel)
-					if debugLevel {
-						otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Reading active requests", zap.String("function", key1.(string)), zap.String("address", key2.(string)), zap.Int("activeRequests", value.activeRequests))
-					}
-					if value.activeRequests == 0 {
-						if debugLevel {
-							otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Function service with no active requests", zap.String("function", key1.(string)), zap.String("address", key2.(string)), zap.Int("activeRequests", value.activeRequests))
-						}
-						vals = append(vals, value.val)
-					}
-				}
-			}
-			resp.allValues = vals
-			req.responseChannel <- resp
-		case setCPUUtilization:
-			if _, ok := c.cache[req.function]; !ok {
-				c.cache[req.function] = make(map[interface{}]*value)
-			}
-			if _, ok := c.cache[req.function][req.address]; ok {
-				c.cache[req.function][req.address].currentCPUUsage = req.cpuUsage
-			}
-		case markAvailable:
-			if _, ok := c.cache[req.function]; ok {
-				if _, ok = c.cache[req.function][req.address]; ok {
-					if c.cache[req.function][req.address].activeRequests > 0 {
-						c.cache[req.function][req.address].activeRequests--
-						if c.logger.Core().Enabled(zap.DebugLevel) {
-							otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Decrease active requests", zap.String("function", req.function.(string)), zap.String("address", req.address.(string)), zap.Int("activeRequests", c.cache[req.function][req.address].activeRequests))
-						}
-					} else {
-						otelUtils.LoggerWithTraceID(req.ctx, c.logger).Error("Invalid request to decrease active requests", zap.String("function", req.function.(string)), zap.String("address", req.address.(string)), zap.Int("activeRequests", c.cache[req.function][req.address].activeRequests))
-					}
-				}
-			}
-		case deleteValue:
-			delete(c.cache[req.function], req.address)
-			req.responseChannel <- resp
-		default:
-			resp.error = ferror.MakeError(ferror.ErrorInvalidArgument,
-				fmt.Sprintf("invalid request type: %v", req.requestType))
-			req.responseChannel <- resp
-		}
-	}
-}
-
-// GetValue returns a value interface with status inActive else return error
-func (c *Cache) GetValue(ctx context.Context, function interface{}, requestsPerPod int) (interface{}, int, error) {
-	respChannel := make(chan *response)
-	c.requestChannel <- &request{
-		ctx:             ctx,
-		requestType:     getValue,
-		function:        function,
-		requestsPerPod:  requestsPerPod,
-		responseChannel: respChannel,
-	}
-	resp := <-respChannel
-	return resp.value, resp.totalActive, resp.error
-}
-
-// ListAvailableValue returns a list of the available function services stored in the Cache
-func (c *Cache) ListAvailableValue() []interface{} {
-	respChannel := make(chan *response)
-	c.requestChannel <- &request{
-		requestType:     listAvailableValue,
-		responseChannel: respChannel,
-	}
-	resp := <-respChannel
-	return resp.allValues
-}
-
-// SetValue marks the value at key [function][address] as active(begin used)
-func (c *Cache) SetValue(ctx context.Context, function, address, value interface{}, cpuLimit resource.Quantity) {
-	respChannel := make(chan *response)
-	c.requestChannel <- &request{
-		ctx:             ctx,
-		requestType:     setValue,
-		function:        function,
-		address:         address,
-		value:           value,
-		cpuUsage:        cpuLimit,
-		responseChannel: respChannel,
-	}
-}
-
-// SetCPUUtilization updates/sets the CPU utilization limit for the pod
-func (c *Cache) SetCPUUtilization(function, address interface{}, cpuUsage resource.Quantity) {
-	c.requestChannel <- &request{
-		requestType:     setCPUUtilization,
-		function:        function,
-		address:         address,
-		cpuUsage:        cpuUsage,
-		responseChannel: make(chan *response),
-	}
-}
-
-// MarkAvailable marks the value at key [function][address] as available
-func (c *Cache) MarkAvailable(function, address interface{}) {
-	respChannel := make(chan *response)
-	c.requestChannel <- &request{
-		requestType:     markAvailable,
-		function:        function,
-		address:         address,
-		responseChannel: respChannel,
-	}
-}
-
-// DeleteValue deletes the value at key composed of [function][address]
-func (c *Cache) DeleteValue(ctx context.Context, function, address interface{}) error {
-	respChannel := make(chan *response)
-	c.requestChannel <- &request{
-		ctx:             ctx,
-		requestType:     deleteValue,
-		function:        function,
-		address:         address,
-		responseChannel: respChannel,
-	}
-	resp := <-respChannel
-	return resp.error
-}

pkg/poolcache/poolcache_test.go

@@ -1,64 +0,0 @@
-package poolcache
-
-import (
-	"context"
-	"log"
-	"testing"
-
-	"k8s.io/apimachinery/pkg/api/resource"
-
-	"github.com/fission/fission/pkg/utils/loggerfactory"
-)
-
-func checkErr(err error) {
-	if err != nil {
-		log.Panicf("err: %v", err)
-	}
-}
-
-func TestPoolCache(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	logger := loggerfactory.GetLogger()
-	c := NewPoolCache(logger)
-
-	c.SetValue(ctx, "func", "ip", "value", resource.MustParse("45m"))
-
-	c.SetValue(ctx, "func2", "ip2", "value2", resource.MustParse("50m"))
-
-	c.SetValue(ctx, "func2", "ip22", "value22", resource.MustParse("33m"))
-
-	checkErr(c.DeleteValue(ctx, "func2", "ip2"))
-
-	cc := c.ListAvailableValue()
-	if len(cc) != 0 {
-		log.Panicf("expected 0 available items")
-	}
-
-	c.MarkAvailable("func", "ip")
-
-	_, active, err := c.GetValue(ctx, "func", 5)
-	if active != 1 {
-		log.Panicln("Expected 1 active, found", active)
-	}
-	checkErr(err)
-
-	checkErr(c.DeleteValue(ctx, "func", "ip"))
-
-	_, _, err = c.GetValue(ctx, "func", 5)
-	if err == nil {
-		log.Panicf("found deleted element")
-	}
-
-	c.SetValue(ctx, "cpulimit", "100", "value", resource.MustParse("3m"))
-	c.SetCPUUtilization("cpulimit", "100", resource.MustParse("4m"))
-
-	_, _, err = c.GetValue(ctx, "cpulimit", 5)
-
-	if err == nil {
-		log.Panicf("received pod address with higher CPU usage than limit")
-	}
-	c.SetCPUUtilization("cpulimit", "100", resource.MustParse("2m"))
-	_, _, err = c.GetValue(ctx, "cpulimit", 5)
-	checkErr(err)
-}

pkg/publisher/publisher.go

@@ -16,13 +16,15 @@ limitations under the License.
 
 package publisher
 
+import "context"
+
 type (
 	// Publisher interface wraps the Publish method that publishes an request
 	// with given "body" and "headers" to given "target"
 	Publisher interface {
-		// Publish an request to a "target". Target's meaning depends on the
+		// Publish a request to a "target". Target's meaning depends on the
 		// publisher: it's a URL in the case of a webhook publisher, or a queue
 		// name in a queue-based publisher such as NATS.
-		Publish(body string, headers map[string]string, target string)
+		Publish(ctx context.Context, body string, headers map[string]string, target string)
 	}
 )

pkg/publisher/publisher_test.go

@@ -0,0 +1,34 @@
+package publisher
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/fission/fission/pkg/utils/loggerfactory"
+	otelUtils "github.com/fission/fission/pkg/utils/otel"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPublisher(t *testing.T) {
+	fnName := "test-fn"
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Equal(t, "/"+fnName, r.URL.Path)
+		assert.Equal(t, "aaa", r.Header.Get("X-Fission-Test"))
+		assert.Contains(t, r.Header, "Traceparent")
+	}))
+
+	ctx := context.Background()
+	logger := loggerfactory.GetLogger()
+	shutdown, err := otelUtils.InitProvider(ctx, logger, fnName)
+	assert.NoError(t, err)
+	if shutdown != nil {
+		defer shutdown(ctx)
+	}
+
+	wp := MakeWebhookPublisher(logger, s.URL)
+	wp.Publish(ctx, "", map[string]string{"X-Fission-Test": "aaa"}, fnName)
+	time.Sleep(time.Second * 1)
+}

pkg/publisher/webhookPublisher.go

@@ -18,12 +18,17 @@ package publisher
 
 import (
 	"bytes"
+	"context"
 	"io"
 	"net/http"
 	"strings"
 	"time"
 
+	otelUtils "github.com/fission/fission/pkg/utils/otel"
+	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
+	"go.opentelemetry.io/otel"
 	"go.uber.org/zap"
+	"golang.org/x/net/context/ctxhttp"
 )
 
 type (
@@ -37,8 +42,10 @@ type (
 		retryDelay time.Duration
 
 		baseURL string
+		timeout time.Duration
 	}
 	publishRequest struct {
+		ctx        context.Context
 		body       string
 		headers    map[string]string
 		target     string
@@ -54,6 +61,8 @@ func MakeWebhookPublisher(logger *zap.Logger, baseURL string) *WebhookPublisher
 		baseURL:        baseURL,
 		requestChannel: make(chan *publishRequest, 32), // buffered channel
 		// TODO make this configurable
+		timeout: 60 * time.Minute,
+		// TODO make this configurable
 		maxRetries: 10,
 		retryDelay: 500 * time.Millisecond,
 	}
@@ -62,9 +71,14 @@ func MakeWebhookPublisher(logger *zap.Logger, baseURL string) *WebhookPublisher
 }
 
 // Publish sends a request to the target with payload having given body and headers
-func (p *WebhookPublisher) Publish(body string, headers map[string]string, target string) {
+func (p *WebhookPublisher) Publish(ctx context.Context, body string, headers map[string]string, target string) {
+	tracer := otel.Tracer("WebhookPublisher")
+	ctx, span := tracer.Start(ctx, "WebhookPublisher/Publish")
+	defer span.End()
+
 	// serializing the request gives user a guarantee that the request is sent in sequence order
 	p.requestChannel <- &publishRequest{
+		ctx:        ctx,
 		body:       body,
 		headers:    headers,
 		target:     target,
@@ -89,7 +103,7 @@ func (p *WebhookPublisher) makeHTTPRequest(r *publishRequest) {
 
 	// log once for this request
 	defer func() {
-		if ce := p.logger.Check(level, msg); ce != nil {
+		if ce := otelUtils.LoggerWithTraceID(r.ctx, p.logger).Check(level, msg); ce != nil {
 			ce.Write(fields...)
 		}
 	}()
@@ -107,7 +121,9 @@ func (p *WebhookPublisher) makeHTTPRequest(r *publishRequest) {
 		req.Header.Set(k, v)
 	}
 	// Make the request
-	resp, err := http.DefaultClient.Do(req)
+	ctx, cancel := context.WithTimeout(r.ctx, p.timeout)
+	defer cancel()
+	resp, err := ctxhttp.Do(ctx, otelhttp.DefaultClient, req)
 	if err != nil {
 		fields = append(fields, zap.Error(err), zap.Any("request", r))
 	} else {

pkg/router/functionHandler.go

@@ -111,11 +111,6 @@ type (
 	}
 )
 
-func init() {
-	// just seeding the random number for getting the canary function
-	rand.Seed(time.Now().UnixNano())
-}
-
 func (w *fakeCloseReadCloser) Close() error {
 	return nil
 }

pkg/timer/timer.go

@@ -17,6 +17,8 @@ limitations under the License.
 package timer
 
 import (
+	"context"
+
 	"github.com/robfig/cron/v3"
 	"go.uber.org/zap"
 
@@ -66,7 +68,7 @@ func (timer *Timer) newCron(t fv1.TimeTrigger) *cron.Cron {
 		// with the addition of multi-tenancy, the users can create functions in any namespace. however,
 		// the triggers can only be created in the same namespace as the function.
 		// so essentially, function namespace = trigger namespace.
-		(*timer.publisher).Publish("", headers, utils.UrlForFunction(t.Spec.FunctionReference.Name, t.Namespace))
+		(*timer.publisher).Publish(context.Background(), "", headers, utils.UrlForFunction(t.Spec.FunctionReference.Name, t.Namespace))
 	})
 	c.Start()
 	timer.logger.Info("started cron for time trigger", zap.String("trigger_name", t.Name), zap.String("trigger_namespace", t.Namespace), zap.String("cron", t.Spec.Cron))

pkg/utils/httpserver/server.go

@@ -4,13 +4,17 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"strings"
 
 	"go.uber.org/zap"
 )
 
 func StartServer(ctx context.Context, log *zap.Logger, svc string, port string, handler http.Handler) {
+	if !strings.Contains(port, ":") {
+		port = fmt.Sprintf(":%s", port)
+	}
 	server := http.Server{
-		Addr:    fmt.Sprintf(":%s", port),
+		Addr:    port,
 		Handler: handler,
 	}
 	l := log.With(zap.String("service", svc), zap.String("addr", server.Addr))

pkg/utils/serviceaccount.go

@@ -145,11 +145,17 @@ func setupSAAndRoleBindings(ctx context.Context, client kubernetes.Interface, lo
 	for _, permission := range ps.permissions {
 		permission.exists, err = checkPermission(ctx, client, SAObj, permission.gvr, permission.verb)
 		if err != nil {
-			//  some error occurred while checking permission
+			//  some error occurred while checking permission, log error as warning message and continue to create new permissions
-			//  now assume permission not exists and will add this permission in rules, insted of return
+			logger.Info(err.Error())
-			logger.Error("error while checking permission", zap.Error(err))
 		}
 		if !permission.exists {
+			logger.Info("creating new permission",
+				zap.String("service_account", SAObj.Name),
+				zap.String("namespace", SAObj.Namespace),
+				zap.String("group", permission.gvr.Group),
+				zap.String("resource", permission.gvr.Resource),
+				zap.String("verb", permission.verb))
+
 			rules = append(rules, rbac.PolicyRule{
 				APIGroups: []string{permission.gvr.Group},
 				Resources: []string{permission.gvr.Resource},

skaffold.yaml

@@ -61,7 +61,7 @@ manifests:
         preUpgradeChecks.imageTag: ""
         priorityClassName: system-cluster-critical
         prometheus.serviceEndpoint: ""
-        repository: index.docker.io
+        repository: ghcr.io
         routerServiceType: LoadBalancer
         runtimePodSpec.enabled: "false"
         serviceMonitor.additionalServiceMonitorLabels.release: prometheus

test/upgrade_test/fission_objects.sh

@@ -5,7 +5,7 @@ ns="fission"
 ROOT=$(pwd)
 PREV_STABLE_VERSION=v1.16.3
 HELM_VARS_PREV_RELEASE="routerServiceType=NodePort,analytics=false"
-HELM_VARS_LATEST_RELEASE="routerServiceType=NodePort,repository=docker.io/library,image=fission-bundle,pullPolicy=IfNotPresent,imageTag=latest,fetcher.image=docker.io/library/fetcher,fetcher.imageTag=latest,postInstallReportImage=reporter,preUpgradeChecks.image=preupgradechecks,preUpgradeChecks.imageTag=latest,analytics=false"
+HELM_VARS_LATEST_RELEASE="routerServiceType=NodePort,repository=ghcr.io,image=fission/fission-bundle,pullPolicy=IfNotPresent,imageTag=latest,fetcher.image=fission/fetcher,fetcher.imageTag=latest,postInstallReportImage=fission/reporter,preUpgradeChecks.image=fission/pre-upgrade-checks,preUpgradeChecks.imageTag=latest,analytics=false"
 
 doit() {
     echo "! $*"