Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0159802497 | ||
|
|
90d781ca2d | ||
|
|
1732d9f4b0 | ||
|
|
8de5a5b0f3 | ||
|
|
35276a503b | ||
|
|
c6c811ea76 | ||
|
|
f2baa17131 | ||
|
|
a34840b0a7 | ||
|
|
06b52e3631 | ||
|
|
19858521fd | ||
|
|
4c4b574d07 | ||
|
|
9b978cf2fd | ||
|
|
ef05e242e4 | ||
|
|
156c1ac2e6 | ||
|
|
ade2daa013 | ||
|
|
6a84cee1d8 | ||
|
|
b95e317a20 |
@@ -87,6 +87,7 @@ jobs:
|
||||
uses: goreleaser/goreleaser-action@7ec5c2b0c6cdda6e8bbb49444bc797dd33d74dd8 # v5.0.0
|
||||
with:
|
||||
install-only: true
|
||||
version: "~> v1"
|
||||
|
||||
- name: Setup Prometheus Stack
|
||||
run: |
|
||||
@@ -229,6 +230,7 @@ jobs:
|
||||
uses: goreleaser/goreleaser-action@7ec5c2b0c6cdda6e8bbb49444bc797dd33d74dd8 # v5.0.0
|
||||
with:
|
||||
install-only: true
|
||||
version: "~> v1"
|
||||
|
||||
- name: Setup Prometheus Stack
|
||||
run: |
|
||||
|
||||
@@ -33,6 +33,7 @@ jobs:
|
||||
uses: goreleaser/goreleaser-action@7ec5c2b0c6cdda6e8bbb49444bc797dd33d74dd8 # v5.0.0
|
||||
with:
|
||||
install-only: true
|
||||
version: "~> v1"
|
||||
|
||||
- name: Kind Clutser
|
||||
uses: helm/kind-action@0025e74a8c7512023d06dc019c617aa3cf561fde # v1.10.0
|
||||
@@ -81,7 +82,7 @@ jobs:
|
||||
- name: Run GoReleaser
|
||||
uses: goreleaser/goreleaser-action@7ec5c2b0c6cdda6e8bbb49444bc797dd33d74dd8 # v5.0.0
|
||||
with:
|
||||
version: latest
|
||||
version: "~> v1"
|
||||
args: release
|
||||
env:
|
||||
COSIGN_PWD: ${{ secrets.COSIGN_PWD }}
|
||||
|
||||
@@ -61,6 +61,7 @@ jobs:
|
||||
uses: goreleaser/goreleaser-action@7ec5c2b0c6cdda6e8bbb49444bc797dd33d74dd8 # v5.0.0
|
||||
with:
|
||||
install-only: true
|
||||
version: "~> v1"
|
||||
|
||||
- name: Setup kubectl & fetch node information
|
||||
run: |
|
||||
|
||||
+10
-99
@@ -88,6 +88,7 @@ dockers:
|
||||
- "ghcr.io/fission/builder:{{ .Tag }}-amd64"
|
||||
dockerfile: cmd/builder/Dockerfile
|
||||
build_flag_templates:
|
||||
- "--label=org.opencontainers.image.description=The builder assists in building the fission function source code for deployment."
|
||||
- "--label=org.opencontainers.image.source={{.GitURL}}"
|
||||
- "--platform=linux/amd64"
|
||||
- "--label=org.opencontainers.image.created={{.Date}}"
|
||||
@@ -103,6 +104,7 @@ dockers:
|
||||
- "ghcr.io/fission/fetcher:{{ .Tag }}-amd64"
|
||||
dockerfile: cmd/fetcher/Dockerfile
|
||||
build_flag_templates:
|
||||
- "--label=org.opencontainers.image.description=Fetcher is a lightweight component used by environment and builder pods. Fetcher helps in fetch and upload of source/deployment packages and specializing environments."
|
||||
- "--label=org.opencontainers.image.source={{.GitURL}}"
|
||||
- "--platform=linux/amd64"
|
||||
- "--label=org.opencontainers.image.created={{.Date}}"
|
||||
@@ -118,6 +120,7 @@ dockers:
|
||||
- "ghcr.io/fission/fission-bundle:{{ .Tag }}-amd64"
|
||||
dockerfile: cmd/fission-bundle/Dockerfile
|
||||
build_flag_templates:
|
||||
- "--label=org.opencontainers.image.description=fission-bundle is a component which is a single binary for all components. Most server side components running on server side are fission-bundle binary wrapped in container and used with different arguments."
|
||||
- "--label=org.opencontainers.image.source={{.GitURL}}"
|
||||
- "--platform=linux/amd64"
|
||||
- "--label=org.opencontainers.image.created={{.Date}}"
|
||||
@@ -133,6 +136,7 @@ dockers:
|
||||
- "ghcr.io/fission/pre-upgrade-checks:{{ .Tag }}-amd64"
|
||||
dockerfile: cmd/preupgradechecks/Dockerfile
|
||||
build_flag_templates:
|
||||
- "--label=org.opencontainers.image.description=Preupgradechecks ensures that Fission is ready for the targeted version upgrade by performing checks beforehand."
|
||||
- "--label=org.opencontainers.image.source={{.GitURL}}"
|
||||
- "--platform=linux/amd64"
|
||||
- "--label=org.opencontainers.image.created={{.Date}}"
|
||||
@@ -148,6 +152,7 @@ dockers:
|
||||
- "ghcr.io/fission/reporter:{{ .Tag }}-amd64"
|
||||
dockerfile: cmd/reporter/Dockerfile
|
||||
build_flag_templates:
|
||||
- "--label=org.opencontainers.image.description=The reporter gathers information that assists in improving fission."
|
||||
- "--label=org.opencontainers.image.source={{.GitURL}}"
|
||||
- "--platform=linux/amd64"
|
||||
- "--label=org.opencontainers.image.created={{.Date}}"
|
||||
@@ -166,6 +171,7 @@ dockers:
|
||||
- "ghcr.io/fission/builder:{{ .Tag }}-arm64"
|
||||
dockerfile: cmd/builder/Dockerfile
|
||||
build_flag_templates:
|
||||
- "--label=org.opencontainers.image.description=The builder assists in building the fission function source code for deployment."
|
||||
- "--label=org.opencontainers.image.source={{.GitURL}}"
|
||||
- "--platform=linux/arm64"
|
||||
- "--label=org.opencontainers.image.created={{.Date}}"
|
||||
@@ -181,6 +187,7 @@ dockers:
|
||||
- "ghcr.io/fission/fetcher:{{ .Tag }}-arm64"
|
||||
dockerfile: cmd/fetcher/Dockerfile
|
||||
build_flag_templates:
|
||||
- "--label=org.opencontainers.image.description=Fetcher is a lightweight component used by environment and builder pods. Fetcher helps in fetch and upload of source/deployment packages and specializing environments."
|
||||
- "--label=org.opencontainers.image.source={{.GitURL}}"
|
||||
- "--platform=linux/arm64"
|
||||
- "--label=org.opencontainers.image.created={{.Date}}"
|
||||
@@ -196,6 +203,7 @@ dockers:
|
||||
- "ghcr.io/fission/fission-bundle:{{ .Tag }}-arm64"
|
||||
dockerfile: cmd/fission-bundle/Dockerfile
|
||||
build_flag_templates:
|
||||
- "--label=org.opencontainers.image.description=fission-bundle is a component which is a single binary for all components. Most server side components running on server side are fission-bundle binary wrapped in container and used with different arguments."
|
||||
- "--label=org.opencontainers.image.source={{.GitURL}}"
|
||||
- "--platform=linux/arm64"
|
||||
- "--label=org.opencontainers.image.created={{.Date}}"
|
||||
@@ -211,6 +219,7 @@ dockers:
|
||||
- "ghcr.io/fission/pre-upgrade-checks:{{ .Tag }}-arm64"
|
||||
dockerfile: cmd/preupgradechecks/Dockerfile
|
||||
build_flag_templates:
|
||||
- "--label=org.opencontainers.image.description=Preupgradechecks ensures that Fission is ready for the targeted version upgrade by performing checks beforehand."
|
||||
- "--label=org.opencontainers.image.source={{.GitURL}}"
|
||||
- "--platform=linux/arm64"
|
||||
- "--label=org.opencontainers.image.created={{.Date}}"
|
||||
@@ -226,191 +235,93 @@ dockers:
|
||||
- "ghcr.io/fission/reporter:{{ .Tag }}-arm64"
|
||||
dockerfile: cmd/reporter/Dockerfile
|
||||
build_flag_templates:
|
||||
- "--label=org.opencontainers.image.description=The reporter gathers information that assists in improving fission."
|
||||
- "--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
|
||||
goarch: arm
|
||||
goarm: 7
|
||||
ids:
|
||||
- builder
|
||||
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}}"
|
||||
- "--label=org.opencontainers.image.version={{.Tag}}"
|
||||
- <<: *docker-armv7
|
||||
ids:
|
||||
- fetcher
|
||||
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
|
||||
- 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
|
||||
- fission/reporter:latest-arm64
|
||||
- fission/reporter:latest-armv7
|
||||
changelog:
|
||||
skip: false
|
||||
archives:
|
||||
|
||||
@@ -40,8 +40,9 @@
|
||||
|
||||
--------------
|
||||
|
||||
Fission is a fast serverless framework for Kubernetes with a focus on
|
||||
developer productivity and high performance.
|
||||
Fission is an open-source, Kubernetes-native serverless framework that simplifies the deployment of functions and applications on Kubernetes. With Fission, developers can easily create and deploy serverless functions that can be triggered by a variety of events, such as HTTP requests, messages from a message queue, or scheduled tasks.
|
||||
|
||||
Fission provides a simple, easy-to-use interface for developers to create serverless functions in their language of choice, without having to worry about the underlying infrastructure. The framework also offers automatic scaling, so functions can scale up or down based on demand, without any additional configuration.
|
||||
|
||||
Fission operates on _just the code_: Docker and Kubernetes are
|
||||
abstracted away under normal operation, though you can use both to
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
apiVersion: v2
|
||||
name: fission-all
|
||||
version: v1.20.2
|
||||
appVersion: v1.20.2
|
||||
version: v1.20.3
|
||||
appVersion: v1.20.3
|
||||
description: Fission is a fast serverless framework for Kubernetes.
|
||||
kubeVersion: ">=1.23.0-0"
|
||||
kubeVersion: ">=1.25.0-0"
|
||||
home: https://fission.io/
|
||||
icon: https://fission.io/images/fission-logo-white.svg
|
||||
sources:
|
||||
|
||||
@@ -354,6 +354,8 @@ rules:
|
||||
# TODO: Kept for future in case preupgrade needs any permissions in the future
|
||||
rules: []
|
||||
{{- end }}
|
||||
# TODO: Currently, router needs ingress related permissions only.
|
||||
# In future if router's permissions are modified then check the configured namespace.
|
||||
{{- define "router-kuberules" }}
|
||||
rules:
|
||||
- apiGroups:
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
{{- 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 }}
|
||||
|
||||
@@ -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.20.2
|
||||
imageTag: v1.20.3
|
||||
|
||||
## pullPolicy represents the pull policy to use for images in the chart.
|
||||
##
|
||||
@@ -106,7 +106,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.20.2
|
||||
imageTag: v1.20.3
|
||||
|
||||
## Fetcher is only for to downloading or uploading archive.
|
||||
## Normally, you don't need to change the value here, unless necessary.
|
||||
@@ -663,7 +663,7 @@ preUpgradeChecks:
|
||||
image: fission/pre-upgrade-checks
|
||||
## pre-install/pre-upgrade checks image version
|
||||
##
|
||||
imageTag: v1.20.2
|
||||
imageTag: v1.20.3
|
||||
|
||||
## Fission post-install/post-upgrade reporting live in this image
|
||||
##
|
||||
@@ -795,28 +795,28 @@ mqt_keda:
|
||||
connector_images:
|
||||
kafka:
|
||||
image: ghcr.io/fission/keda-kafka-http-connector
|
||||
tag: v0.13
|
||||
tag: v0.15
|
||||
rabbitmq:
|
||||
image: ghcr.io/fission/keda-rabbitmq-http-connector
|
||||
tag: v0.11
|
||||
tag: v0.13
|
||||
awskinesis:
|
||||
image: ghcr.io/fission/keda-aws-kinesis-http-connector
|
||||
tag: v0.11
|
||||
tag: v0.13
|
||||
aws_sqs:
|
||||
image: ghcr.io/fission/keda-aws-sqs-http-connector
|
||||
tag: v0.12
|
||||
tag: v0.14
|
||||
nats_steaming:
|
||||
image: ghcr.io/fission/keda-nats-streaming-http-connector
|
||||
tag: v0.14
|
||||
tag: v0.16
|
||||
nats_jetstream:
|
||||
image: ghcr.io/fission/keda-nats-jetstream-http-connector
|
||||
tag: v0.5
|
||||
tag: v0.7
|
||||
gcp_pubsub:
|
||||
image: ghcr.io/fission/keda-gcp-pubsub-http-connector
|
||||
tag: v0.7
|
||||
tag: v0.9
|
||||
redis:
|
||||
image: ghcr.io/fission/keda-redis-http-connector
|
||||
tag: v0.4
|
||||
tag: v0.6
|
||||
|
||||
## Pod resources as:
|
||||
## resources:
|
||||
|
||||
@@ -32,6 +32,7 @@ func Run(ctx context.Context, logger *zap.Logger, mgr manager.Interface, shareVo
|
||||
builder := builder.MakeBuilder(logger, shareVolume)
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/", builder.Handler)
|
||||
mux.HandleFunc("/clean", builder.Clean)
|
||||
mux.HandleFunc("/version", builder.VersionHandler)
|
||||
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
|
||||
@@ -20,7 +20,7 @@ require (
|
||||
github.com/gorilla/mux v1.8.1
|
||||
github.com/graymeta/stow v0.2.8
|
||||
github.com/hashicorp/go-multierror v1.1.1
|
||||
github.com/hashicorp/go-retryablehttp v0.7.5
|
||||
github.com/hashicorp/go-retryablehttp v0.7.7
|
||||
github.com/influxdata/influxdb v1.11.5
|
||||
github.com/mholt/archiver/v3 v3.5.1
|
||||
github.com/minio/minio-go v6.0.14+incompatible
|
||||
@@ -163,15 +163,15 @@ require (
|
||||
go.uber.org/multierr v1.11.0 // indirect
|
||||
golang.org/x/crypto v0.23.0 // indirect
|
||||
golang.org/x/exp v0.0.0-20231214170342-aacd6d4b4611 // indirect
|
||||
golang.org/x/image v0.10.0 // indirect
|
||||
golang.org/x/image v0.18.0 // indirect
|
||||
golang.org/x/mod v0.17.0 // indirect
|
||||
golang.org/x/oauth2 v0.18.0 // indirect
|
||||
golang.org/x/sync v0.7.0 // indirect
|
||||
golang.org/x/sys v0.20.0 // indirect
|
||||
golang.org/x/term v0.20.0 // indirect
|
||||
golang.org/x/text v0.15.0 // indirect
|
||||
golang.org/x/text v0.16.0 // indirect
|
||||
golang.org/x/time v0.3.0 // indirect
|
||||
golang.org/x/tools v0.20.0 // indirect
|
||||
golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect
|
||||
golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 // indirect
|
||||
gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
|
||||
google.golang.org/appengine v1.6.8 // indirect
|
||||
|
||||
@@ -238,13 +238,13 @@ github.com/hashicorp/errwrap v1.0.0 h1:hLrqtEDnRye3+sgx6z4qVLNuviH3MR5aQ0ykNJa/U
|
||||
github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
|
||||
github.com/hashicorp/go-cleanhttp v0.5.2 h1:035FKYIWjmULyFRBKPs8TBQoi0x6d9G4xc9neXJWAZQ=
|
||||
github.com/hashicorp/go-cleanhttp v0.5.2/go.mod h1:kO/YDlP8L1346E6Sodw+PrpBSV4/SoxCXGY6BqNFT48=
|
||||
github.com/hashicorp/go-hclog v0.9.2 h1:CG6TE5H9/JXsFWJCfoIVpKFIkFe6ysEuHirp4DxCsHI=
|
||||
github.com/hashicorp/go-hclog v0.9.2/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ=
|
||||
github.com/hashicorp/go-hclog v1.6.3 h1:Qr2kF+eVWjTiYmU7Y31tYlP1h0q/X3Nl3tPGdaB11/k=
|
||||
github.com/hashicorp/go-hclog v1.6.3/go.mod h1:W4Qnvbt70Wk/zYJryRzDRU/4r0kIg0PVHBcfoyhpF5M=
|
||||
github.com/hashicorp/go-multierror v1.0.0/go.mod h1:dHtQlpGsu+cZNNAkkCN/P3hoUDHhCYQXV3UM06sGGrk=
|
||||
github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo=
|
||||
github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM=
|
||||
github.com/hashicorp/go-retryablehttp v0.7.5 h1:bJj+Pj19UZMIweq/iie+1u5YCdGrnxCT9yvm0e+Nd5M=
|
||||
github.com/hashicorp/go-retryablehttp v0.7.5/go.mod h1:Jy/gPYAdjqffZ/yFGCFV2doI5wjtH1ewM9u8iYVjtX8=
|
||||
github.com/hashicorp/go-retryablehttp v0.7.7 h1:C8hUCYzor8PIfXHa4UrZkU4VvK8o9ISHxT2Q8+VepXU=
|
||||
github.com/hashicorp/go-retryablehttp v0.7.7/go.mod h1:pkQpWZeYWskR+D1tR2O5OcBFOxfA7DoAO6xtkuQnHTk=
|
||||
github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
|
||||
github.com/hashicorp/go-uuid v1.0.3 h1:2gKiV6YVmrJ1i2CKKa9obLvRieoRGviZFL26PcT/Co8=
|
||||
github.com/hashicorp/go-uuid v1.0.3/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
|
||||
@@ -518,8 +518,8 @@ golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v
|
||||
golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
|
||||
golang.org/x/exp v0.0.0-20231214170342-aacd6d4b4611 h1:qCEDpW1G+vcj3Y7Fy52pEM1AWm3abj8WimGYejI3SC4=
|
||||
golang.org/x/exp v0.0.0-20231214170342-aacd6d4b4611/go.mod h1:iRJReGqOEeBhDZGkGbynYwcHlctCvnjTYIamk7uXpHI=
|
||||
golang.org/x/image v0.10.0 h1:gXjUUtwtx5yOE0VKWq1CH4IJAClq4UGgUA3i+rpON9M=
|
||||
golang.org/x/image v0.10.0/go.mod h1:jtrku+n79PfroUbvDdeUWMAI+heR786BofxrbiSF+J0=
|
||||
golang.org/x/image v0.18.0 h1:jGzIakQa/ZXI1I0Fxvaa9W7yP25TqT6cHIHn+6CqvSQ=
|
||||
golang.org/x/image v0.18.0/go.mod h1:4yyo5vMFQjVjUcVk4jEQcU9MGy/rulF5WvUILseCM2E=
|
||||
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=
|
||||
@@ -615,9 +615,8 @@ golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
|
||||
golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
|
||||
golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
|
||||
golang.org/x/text v0.8.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
|
||||
golang.org/x/text v0.11.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
|
||||
golang.org/x/text v0.15.0 h1:h1V/4gjBv8v9cjcR6+AR5+/cIYK5N/WAgiv4xlsEtAk=
|
||||
golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
|
||||
golang.org/x/text v0.16.0 h1:a94ExnEXNtEwYLGJSIUxnWoxoRz/ZcCsV63ROupILh4=
|
||||
golang.org/x/text v0.16.0/go.mod h1:GhwF1Be+LQoKShO3cGOHzqOgRrGaYc9AvblQOmPVHnI=
|
||||
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.3.0 h1:rg5rLMjNzMS1RkNLzCG38eapWhnYLFYXDXj2gOlr8j4=
|
||||
@@ -634,8 +633,8 @@ golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roY
|
||||
golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
|
||||
golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
|
||||
golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
|
||||
golang.org/x/tools v0.20.0 h1:hz/CVckiOxybQvFw6h7b/q80NTr9IUQb4s1IIzW7KNY=
|
||||
golang.org/x/tools v0.20.0/go.mod h1:WvitBU7JJf6A4jOdg4S1tviW9bhUxkgeCui/0JHctQg=
|
||||
golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d h1:vU5i/LfpvrRCpgM/VPfJLg5KjxD3E+hfT1SH+d9zLwg=
|
||||
golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk=
|
||||
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
|
||||
@@ -35,6 +35,7 @@ import (
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/fission/fission/pkg/info"
|
||||
"github.com/fission/fission/pkg/utils"
|
||||
otelUtils "github.com/fission/fission/pkg/utils/otel"
|
||||
)
|
||||
|
||||
@@ -154,6 +155,38 @@ func (builder *Builder) Handler(w http.ResponseWriter, r *http.Request) {
|
||||
builder.reply(r.Context(), w, deployPkgFilename, buildLogs, http.StatusOK)
|
||||
}
|
||||
|
||||
func (builder *Builder) Clean(w http.ResponseWriter, r *http.Request) {
|
||||
logger := otelUtils.LoggerWithTraceID(r.Context(), builder.logger)
|
||||
|
||||
if r.Method != "DELETE" {
|
||||
e := "method not allowed"
|
||||
logger.Error(e, zap.String("http_method", r.Method))
|
||||
builder.reply(r.Context(), w, "", fmt.Sprintf("%s: %s", e, r.Method), http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
|
||||
startTime := time.Now()
|
||||
defer func() {
|
||||
elapsed := time.Since(startTime)
|
||||
logger.Info("clean request complete", zap.Duration("elapsed_time", elapsed))
|
||||
}()
|
||||
|
||||
srcPkgFilename := r.URL.Query().Get("name")
|
||||
srcPkgPath := filepath.Join(builder.sharedVolumePath, srcPkgFilename)
|
||||
|
||||
logger.Info("builder received clean request", zap.Any("source_package", srcPkgFilename))
|
||||
|
||||
err := utils.DeleteOldPackages(srcPkgPath, envSrcPkg)
|
||||
if err != nil {
|
||||
e := "error deleting src package after build"
|
||||
logger.Error(e, zap.Error(err))
|
||||
builder.reply(r.Context(), w, srcPkgFilename, "", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
builder.reply(r.Context(), w, srcPkgFilename, "", http.StatusOK)
|
||||
}
|
||||
|
||||
func (builder *Builder) reply(ctx context.Context, w http.ResponseWriter, pkgFilename string, buildLogs string, statusCode int) {
|
||||
logger := otelUtils.LoggerWithTraceID(ctx, builder.logger)
|
||||
resp := PackageBuildResponse{
|
||||
|
||||
@@ -168,4 +168,50 @@ func TestBuilder(t *testing.T) {
|
||||
})
|
||||
}
|
||||
})
|
||||
|
||||
// Test CleanHandler
|
||||
t.Run("CleanHandler", func(t *testing.T) {
|
||||
for _, test := range []struct {
|
||||
name string
|
||||
srcPkgFilename string
|
||||
handler func(w http.ResponseWriter, r *http.Request)
|
||||
status int
|
||||
}{
|
||||
{
|
||||
name: "should fail deleting src pkg: invalid shared volume path",
|
||||
srcPkgFilename: "test2",
|
||||
handler: func(w http.ResponseWriter, r *http.Request) {
|
||||
builder.Clean(w, r)
|
||||
},
|
||||
status: http.StatusInternalServerError,
|
||||
},
|
||||
{
|
||||
name: "should fail deleting src pkg: method not allowed",
|
||||
srcPkgFilename: "test3",
|
||||
handler: func(w http.ResponseWriter, r *http.Request) {
|
||||
builder.Handler(w, r)
|
||||
},
|
||||
status: http.StatusMethodNotAllowed,
|
||||
},
|
||||
} {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
_, err := os.MkdirTemp(dir, test.srcPkgFilename)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
srcFile, err := os.Create(dir + "/" + test.srcPkgFilename)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer srcFile.Close()
|
||||
w := httptest.NewRecorder()
|
||||
r := httptest.NewRequest(http.MethodDelete, "/clean/"+test.srcPkgFilename, http.NoBody)
|
||||
test.handler(w, r)
|
||||
resp := w.Result()
|
||||
if resp.StatusCode != test.status {
|
||||
t.Errorf("expected status code %d, got %d", test.status, resp.StatusCode)
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -21,6 +21,7 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"github.com/hashicorp/go-retryablehttp"
|
||||
@@ -37,6 +38,7 @@ import (
|
||||
type (
|
||||
ClientInterface interface {
|
||||
Build(context.Context, *builder.PackageBuildRequest) (*builder.PackageBuildResponse, error)
|
||||
Clean(context.Context, string) error
|
||||
}
|
||||
|
||||
client struct {
|
||||
@@ -57,6 +59,10 @@ func MakeClient(logger *zap.Logger, builderUrl string) ClientInterface {
|
||||
}
|
||||
}
|
||||
|
||||
func (c *client) getCleanUrl(srcPkgFilename string) string {
|
||||
return c.url + "/clean" + "?name=" + srcPkgFilename
|
||||
}
|
||||
|
||||
func (c *client) Build(ctx context.Context, req *builder.PackageBuildRequest) (*builder.PackageBuildResponse, error) {
|
||||
logger := otelUtils.LoggerWithTraceID(ctx, c.logger)
|
||||
|
||||
@@ -86,3 +92,29 @@ func (c *client) Build(ctx context.Context, req *builder.PackageBuildRequest) (*
|
||||
|
||||
return &pkgBuildResp, ferror.MakeErrorFromHTTP(resp)
|
||||
}
|
||||
|
||||
func (c *client) Clean(ctx context.Context, srcPkgFilename string) error {
|
||||
logger := otelUtils.LoggerWithTraceID(ctx, c.logger)
|
||||
|
||||
req, err := http.NewRequest(http.MethodDelete, c.getCleanUrl(srcPkgFilename), http.NoBody)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to create http request for clean api")
|
||||
}
|
||||
|
||||
resp, err := ctxhttp.Do(ctx, c.httpClient.StandardClient(), req)
|
||||
if err != nil {
|
||||
logger.Error("error sending clean request", zap.Error(err))
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode == http.StatusMethodNotAllowed {
|
||||
return nil
|
||||
}
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return ferror.MakeErrorFromHTTP(resp)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -60,6 +60,15 @@ func buildPackage(ctx context.Context, logger *zap.Logger, fissionClient version
|
||||
fetcherC := fetcherClient.MakeClient(logger, fmt.Sprintf("http://%v:8000", svcName))
|
||||
builderC := builderClient.MakeClient(logger, fmt.Sprintf("http://%v:8001", svcName))
|
||||
|
||||
defer func() {
|
||||
logger.Info("cleaning src pkg from builder storage", zap.String("source_package", srcPkgFilename))
|
||||
errC := cleanPackage(ctx, builderC, srcPkgFilename)
|
||||
if errC != nil {
|
||||
m := "error cleaning src pkg from builder storage"
|
||||
logger.Error(m, zap.Error(errC))
|
||||
}
|
||||
}()
|
||||
|
||||
fetchReq := &fetcher.FunctionFetchRequest{
|
||||
FetchType: fv1.FETCH_SOURCE,
|
||||
Package: pkg.ObjectMeta,
|
||||
@@ -121,6 +130,15 @@ func buildPackage(ctx context.Context, logger *zap.Logger, fissionClient version
|
||||
return uploadResp, buildResp.BuildLogs, nil
|
||||
}
|
||||
|
||||
func cleanPackage(ctx context.Context, builderClient builderClient.ClientInterface, srcPkgFileName string) error {
|
||||
err := builderClient.Clean(ctx, srcPkgFileName)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func updatePackage(ctx context.Context, logger *zap.Logger, fissionClient versioned.Interface,
|
||||
pkg *fv1.Package, status fv1.BuildStatus, buildLogs string,
|
||||
uploadResp *fetcher.ArchiveUploadResponse) (*fv1.Package, error) {
|
||||
|
||||
@@ -28,6 +28,7 @@ import (
|
||||
apiv1 "k8s.io/api/core/v1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/labels"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/types"
|
||||
"k8s.io/apimachinery/pkg/util/intstr"
|
||||
"k8s.io/client-go/kubernetes"
|
||||
@@ -329,6 +330,13 @@ func (envw *environmentWatcher) createBuilderService(ctx context.Context, env *f
|
||||
Namespace: ns,
|
||||
Name: name,
|
||||
Labels: sel,
|
||||
OwnerReferences: []metav1.OwnerReference{
|
||||
*metav1.NewControllerRef(env, schema.GroupVersionKind{
|
||||
Group: "fission.io",
|
||||
Version: "v1",
|
||||
Kind: "Environment",
|
||||
}),
|
||||
},
|
||||
},
|
||||
Spec: apiv1.ServiceSpec{
|
||||
Selector: sel,
|
||||
@@ -440,6 +448,13 @@ func (envw *environmentWatcher) createBuilderDeployment(ctx context.Context, env
|
||||
Namespace: ns,
|
||||
Name: name,
|
||||
Labels: sel,
|
||||
OwnerReferences: []metav1.OwnerReference{
|
||||
*metav1.NewControllerRef(env, schema.GroupVersionKind{
|
||||
Group: "fission.io",
|
||||
Version: "v1",
|
||||
Kind: "Environment",
|
||||
}),
|
||||
},
|
||||
},
|
||||
Spec: appsv1.DeploymentSpec{
|
||||
Replicas: &replicas,
|
||||
|
||||
@@ -421,7 +421,7 @@ func (caaf *Container) fnCreate(ctx context.Context, fn *fv1.Function) (*fscache
|
||||
return nil, fmt.Errorf("error creating deployment %s: %w", objName, err)
|
||||
}
|
||||
|
||||
hpa, err := caaf.hpaops.CreateOrGetHpa(ctx, objName, &fn.Spec.InvokeStrategy.ExecutionStrategy, depl, deployLabels, deployAnnotations)
|
||||
hpa, err := caaf.hpaops.CreateOrGetHpa(ctx, fn, objName, &fn.Spec.InvokeStrategy.ExecutionStrategy, depl, deployLabels, deployAnnotations)
|
||||
if err != nil {
|
||||
caaf.logger.Error("error creating HPA", zap.Error(err), zap.String("hpa", objName))
|
||||
go cleanupFunc(ns, objName)
|
||||
|
||||
@@ -27,6 +27,7 @@ import (
|
||||
apiv1 "k8s.io/api/core/v1"
|
||||
k8s_err "k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/util/intstr"
|
||||
|
||||
fv1 "github.com/fission/fission/pkg/apis/core/v1"
|
||||
@@ -86,6 +87,7 @@ func (cn *Container) createOrGetDeployment(ctx context.Context, fn *fv1.Function
|
||||
if existingDepl.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != cn.instanceID {
|
||||
existingDepl.Annotations = deployment.Annotations
|
||||
existingDepl.Labels = deployment.Labels
|
||||
existingDepl.OwnerReferences = deployment.OwnerReferences
|
||||
existingDepl.Spec.Template.Spec.Containers = deployment.Spec.Template.Spec.Containers
|
||||
existingDepl.Spec.Template.Spec.ServiceAccountName = deployment.Spec.Template.Spec.ServiceAccountName
|
||||
existingDepl.Spec.Template.Spec.TerminationGracePeriodSeconds = deployment.Spec.Template.Spec.TerminationGracePeriodSeconds
|
||||
@@ -268,6 +270,13 @@ func (cn *Container) getDeploymentSpec(ctx context.Context, fn *fv1.Function, ta
|
||||
Name: deployName,
|
||||
Labels: deployLabels,
|
||||
Annotations: deployAnnotations,
|
||||
OwnerReferences: []metav1.OwnerReference{
|
||||
*metav1.NewControllerRef(fn, schema.GroupVersionKind{
|
||||
Group: "fission.io",
|
||||
Version: "v1",
|
||||
Kind: "Function",
|
||||
}),
|
||||
},
|
||||
},
|
||||
Spec: appsv1.DeploymentSpec{
|
||||
Replicas: &replicas,
|
||||
|
||||
@@ -24,6 +24,7 @@ import (
|
||||
apiv1 "k8s.io/api/core/v1"
|
||||
k8s_err "k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/util/intstr"
|
||||
|
||||
fv1 "github.com/fission/fission/pkg/apis/core/v1"
|
||||
@@ -54,6 +55,13 @@ func (cn *Container) createOrGetSvc(ctx context.Context, fn *fv1.Function, deplo
|
||||
Name: svcName,
|
||||
Labels: deployLabels,
|
||||
Annotations: deployAnnotations,
|
||||
OwnerReferences: []metav1.OwnerReference{
|
||||
*metav1.NewControllerRef(fn, schema.GroupVersionKind{
|
||||
Group: "fission.io",
|
||||
Version: "v1",
|
||||
Kind: "Function",
|
||||
}),
|
||||
},
|
||||
},
|
||||
Spec: apiv1.ServiceSpec{
|
||||
Ports: []apiv1.ServicePort{
|
||||
@@ -74,6 +82,7 @@ func (cn *Container) createOrGetSvc(ctx context.Context, fn *fv1.Function, deplo
|
||||
if existingSvc.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != cn.instanceID {
|
||||
existingSvc.Annotations = service.Annotations
|
||||
existingSvc.Labels = service.Labels
|
||||
existingSvc.OwnerReferences = service.OwnerReferences
|
||||
existingSvc.Spec.Ports = service.Spec.Ports
|
||||
existingSvc.Spec.Selector = service.Spec.Selector
|
||||
existingSvc.Spec.Type = service.Spec.Type
|
||||
|
||||
@@ -29,6 +29,7 @@ import (
|
||||
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/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/util/intstr"
|
||||
"k8s.io/client-go/kubernetes"
|
||||
|
||||
@@ -61,6 +62,7 @@ func (deploy *NewDeploy) createOrGetDeployment(ctx context.Context, fn *fv1.Func
|
||||
if existingDepl.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != deploy.instanceID {
|
||||
existingDepl.Annotations = deployment.Annotations
|
||||
existingDepl.Labels = deployment.Labels
|
||||
existingDepl.OwnerReferences = deployment.OwnerReferences
|
||||
existingDepl.Spec.Template.Spec.Containers = deployment.Spec.Template.Spec.Containers
|
||||
existingDepl.Spec.Template.Spec.ServiceAccountName = deployment.Spec.Template.Spec.ServiceAccountName
|
||||
existingDepl.Spec.Template.Spec.TerminationGracePeriodSeconds = deployment.Spec.Template.Spec.TerminationGracePeriodSeconds
|
||||
@@ -249,6 +251,13 @@ func (deploy *NewDeploy) getDeploymentSpec(ctx context.Context, fn *fv1.Function
|
||||
Name: deployName,
|
||||
Labels: deployLabels,
|
||||
Annotations: deployAnnotations,
|
||||
OwnerReferences: []metav1.OwnerReference{
|
||||
*metav1.NewControllerRef(fn, schema.GroupVersionKind{
|
||||
Group: "fission.io",
|
||||
Version: "v1",
|
||||
Kind: "Function",
|
||||
}),
|
||||
},
|
||||
},
|
||||
Spec: appsv1.DeploymentSpec{
|
||||
Replicas: &replicas,
|
||||
@@ -323,13 +332,20 @@ func (deploy *NewDeploy) getResources(env *fv1.Environment, fn *fv1.Function) ap
|
||||
return resources
|
||||
}
|
||||
|
||||
func (deploy *NewDeploy) createOrGetSvc(ctx context.Context, deployLabels map[string]string, deployAnnotations map[string]string, svcName string, svcNamespace string) (*apiv1.Service, error) {
|
||||
func (deploy *NewDeploy) createOrGetSvc(ctx context.Context, fn *fv1.Function, deployLabels map[string]string, deployAnnotations map[string]string, svcName string, svcNamespace string) (*apiv1.Service, error) {
|
||||
logger := otelUtils.LoggerWithTraceID(ctx, deploy.logger)
|
||||
service := &apiv1.Service{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: svcName,
|
||||
Labels: deployLabels,
|
||||
Annotations: deployAnnotations,
|
||||
OwnerReferences: []metav1.OwnerReference{
|
||||
*metav1.NewControllerRef(fn, schema.GroupVersionKind{
|
||||
Group: "fission.io",
|
||||
Version: "v1",
|
||||
Kind: "Function",
|
||||
}),
|
||||
},
|
||||
},
|
||||
Spec: apiv1.ServiceSpec{
|
||||
Ports: []apiv1.ServicePort{
|
||||
@@ -351,6 +367,7 @@ func (deploy *NewDeploy) createOrGetSvc(ctx context.Context, deployLabels map[st
|
||||
if existingSvc.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != deploy.instanceID {
|
||||
existingSvc.Annotations = service.Annotations
|
||||
existingSvc.Labels = service.Labels
|
||||
existingSvc.OwnerReferences = service.OwnerReferences
|
||||
existingSvc.Spec.Ports = service.Spec.Ports
|
||||
existingSvc.Spec.Selector = service.Spec.Selector
|
||||
existingSvc.Spec.Type = service.Spec.Type
|
||||
|
||||
@@ -451,7 +451,7 @@ func (deploy *NewDeploy) fnCreate(ctx context.Context, fn *fv1.Function) (*fscac
|
||||
// Since newdeploy waits for pods of deployment to be ready,
|
||||
// change the order of kubeObject creation (create service first,
|
||||
// then deployment) to take advantage of waiting time.
|
||||
svc, err := deploy.createOrGetSvc(ctx, deployLabels, deployAnnotations, objName, ns)
|
||||
svc, err := deploy.createOrGetSvc(ctx, fn, deployLabels, deployAnnotations, objName, ns)
|
||||
if err != nil {
|
||||
deploy.logger.Error("error creating service", zap.Error(err), zap.String("service", objName))
|
||||
go cleanupFunc(context.Background(), ns, objName)
|
||||
@@ -466,7 +466,7 @@ func (deploy *NewDeploy) fnCreate(ctx context.Context, fn *fv1.Function) (*fscac
|
||||
return nil, fmt.Errorf("error creating deployment %s: %w", objName, err)
|
||||
}
|
||||
|
||||
hpa, err := deploy.hpaops.CreateOrGetHpa(ctx, objName, &fn.Spec.InvokeStrategy.ExecutionStrategy, depl, deployLabels, deployAnnotations)
|
||||
hpa, err := deploy.hpaops.CreateOrGetHpa(ctx, fn, objName, &fn.Spec.InvokeStrategy.ExecutionStrategy, depl, deployLabels, deployAnnotations)
|
||||
if err != nil {
|
||||
deploy.logger.Error("error creating HPA", zap.Error(err), zap.String("hpa", objName))
|
||||
go cleanupFunc(context.Background(), ns, objName)
|
||||
|
||||
@@ -26,6 +26,7 @@ import (
|
||||
apiv1 "k8s.io/api/core/v1"
|
||||
k8sErrs "k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
|
||||
fv1 "github.com/fission/fission/pkg/apis/core/v1"
|
||||
"github.com/fission/fission/pkg/executor/util"
|
||||
@@ -62,6 +63,13 @@ func (gp *GenericPool) genDeploymentMeta(env *fv1.Environment) metav1.ObjectMeta
|
||||
Name: getPoolName(env),
|
||||
Labels: deployLabels,
|
||||
Annotations: deployAnnotations,
|
||||
OwnerReferences: []metav1.OwnerReference{
|
||||
*metav1.NewControllerRef(env, schema.GroupVersionKind{
|
||||
Group: "fission.io",
|
||||
Version: "v1",
|
||||
Kind: "Environment",
|
||||
}),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -25,6 +25,7 @@ import (
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
k8s_err "k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/client-go/kubernetes"
|
||||
|
||||
fv1 "github.com/fission/fission/pkg/apis/core/v1"
|
||||
@@ -72,7 +73,7 @@ func getScaleTargetRef(deployment *appsv1.Deployment) asv2.CrossVersionObjectRef
|
||||
}
|
||||
}
|
||||
|
||||
func (hpaops *HpaOperations) CreateOrGetHpa(ctx context.Context, hpaName string, execStrategy *fv1.ExecutionStrategy,
|
||||
func (hpaops *HpaOperations) CreateOrGetHpa(ctx context.Context, fn *fv1.Function, hpaName string, execStrategy *fv1.ExecutionStrategy,
|
||||
depl *appsv1.Deployment, deployLabels map[string]string, deployAnnotations map[string]string) (*asv2.HorizontalPodAutoscaler, error) {
|
||||
|
||||
if depl == nil {
|
||||
@@ -103,6 +104,13 @@ func (hpaops *HpaOperations) CreateOrGetHpa(ctx context.Context, hpaName string,
|
||||
Name: hpaName,
|
||||
Labels: deployLabels,
|
||||
Annotations: deployAnnotations,
|
||||
OwnerReferences: []metav1.OwnerReference{
|
||||
*metav1.NewControllerRef(fn, schema.GroupVersionKind{
|
||||
Group: "fission.io",
|
||||
Version: "v1",
|
||||
Kind: "Function",
|
||||
}),
|
||||
},
|
||||
},
|
||||
Spec: asv2.HorizontalPodAutoscalerSpec{
|
||||
ScaleTargetRef: getScaleTargetRef(depl),
|
||||
@@ -119,6 +127,7 @@ func (hpaops *HpaOperations) CreateOrGetHpa(ctx context.Context, hpaName string,
|
||||
if existingHpa.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != hpaops.instanceID {
|
||||
existingHpa.Annotations = hpa.Annotations
|
||||
existingHpa.Labels = hpa.Labels
|
||||
existingHpa.OwnerReferences = hpa.OwnerReferences
|
||||
existingHpa.Spec = hpa.Spec
|
||||
existingHpa, err = hpaops.kubernetesClient.AutoscalingV2().HorizontalPodAutoscalers(depl.ObjectMeta.Namespace).Update(ctx, existingHpa, metav1.UpdateOptions{})
|
||||
if err != nil {
|
||||
|
||||
@@ -25,6 +25,7 @@ import (
|
||||
asv2 "k8s.io/api/autoscaling/v2"
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/util/uuid"
|
||||
"k8s.io/client-go/kubernetes/fake"
|
||||
|
||||
fv1 "github.com/fission/fission/pkg/apis/core/v1"
|
||||
@@ -67,7 +68,14 @@ func TestHpaOps(t *testing.T) {
|
||||
"test-annotation": "test-annotation-value",
|
||||
}
|
||||
// Test CreateHPA
|
||||
hpa, err := hpaops.CreateOrGetHpa(ctx, "test-hpa",
|
||||
hpa, err := hpaops.CreateOrGetHpa(ctx,
|
||||
&fv1.Function{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "test-fn",
|
||||
UID: uuid.NewUUID(),
|
||||
},
|
||||
},
|
||||
"test-hpa",
|
||||
&fv1.ExecutionStrategy{
|
||||
ExecutorType: fv1.ExecutorTypeNewdeploy,
|
||||
MinScale: 1,
|
||||
|
||||
+15
-2
@@ -525,6 +525,14 @@ func (fetcher *Fetcher) UploadHandler(w http.ResponseWriter, r *http.Request) {
|
||||
srcFilepath := filepath.Join(fetcher.sharedVolumePath, req.Filename)
|
||||
dstFilepath := filepath.Join(fetcher.sharedVolumePath, zipFilename)
|
||||
|
||||
defer func() {
|
||||
errC := utils.DeleteOldPackages(srcFilepath, "DEPLOY_PKG")
|
||||
if errC != nil {
|
||||
m := "error deleting deploy package after upload"
|
||||
logger.Error(m, zap.Error(errC))
|
||||
}
|
||||
}()
|
||||
|
||||
if req.ArchivePackage {
|
||||
err = fetcher.archive(srcFilepath, dstFilepath)
|
||||
if err != nil {
|
||||
@@ -584,6 +592,7 @@ func (fetcher *Fetcher) UploadHandler(w http.ResponseWriter, r *http.Request) {
|
||||
logger.Error(e, zap.Error(err))
|
||||
http.Error(w, fmt.Sprintf("%s: %v", e, err), http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func (fetcher *Fetcher) rename(src string, dst string) error {
|
||||
@@ -611,12 +620,16 @@ func (fetcher *Fetcher) archive(src string, dst string) error {
|
||||
} else {
|
||||
files = append(files, src)
|
||||
}
|
||||
return archiver.DefaultZip.Archive(files, dst)
|
||||
zip := archiver.NewZip()
|
||||
defer zip.Close()
|
||||
return zip.Archive(files, dst)
|
||||
}
|
||||
|
||||
// unarchive is a function that unzips a zip file to destination
|
||||
func (fetcher *Fetcher) unarchive(src string, dst string) error {
|
||||
err := archiver.DefaultZip.Unarchive(src, dst)
|
||||
zip := archiver.NewZip()
|
||||
defer zip.Close()
|
||||
err := zip.Unarchive(src, dst)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to unzip file: %w", err)
|
||||
}
|
||||
|
||||
@@ -36,7 +36,7 @@ func (opts *CheckSubCommand) do(input cli.Input) error {
|
||||
|
||||
userProvidedNS, _, err := opts.GetResourceNamespace(input, flagkey.Namespace)
|
||||
if err != nil {
|
||||
return fmt.Errorf("error retrieving user provided namespace: %w", err)
|
||||
return fmt.Errorf("error retrieving user provided namespace information: %w", err)
|
||||
}
|
||||
|
||||
if input.IsSet(flagkey.PreCheckOnly) {
|
||||
|
||||
@@ -98,6 +98,11 @@ func (opts *CreateSubCommand) complete(input cli.Input) error {
|
||||
|
||||
fnIdleTimeout := input.Int(flagkey.FnIdleTimeout)
|
||||
|
||||
err = checkExecutorPoolManager(input, fv1.ExecutorTypePoolmgr)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
fnConcurrency := DEFAULT_CONCURRENCY
|
||||
if input.IsSet(flagkey.FnConcurrency) {
|
||||
fnConcurrency = input.Int(flagkey.FnConcurrency)
|
||||
@@ -447,6 +452,33 @@ func getInvokeStrategy(input cli.Input, existingInvokeStrategy *fv1.InvokeStrate
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Show warning when --con, --rpp and --yolo flags are used with executortype other than `poolmgr`.
|
||||
// These flags are specifically introduced for executortype `poolmgr`.
|
||||
func checkExecutorPoolManager(input cli.Input, existingExecutorType fv1.ExecutorType) error {
|
||||
var isNotPoolManager bool
|
||||
if input.IsSet(flagkey.EnvExecutorType) {
|
||||
executorType, err := getExecutorType(input)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
isNotPoolManager = (string(executorType) != string(fv1.ExecutorTypePoolmgr))
|
||||
} else {
|
||||
isNotPoolManager = (string(existingExecutorType) != string(fv1.ExecutorTypePoolmgr))
|
||||
}
|
||||
|
||||
if input.IsSet(flagkey.FnConcurrency) && isNotPoolManager {
|
||||
console.Warn("--concurrency is only valid for executortype; `poolmgr`. Check `fission function create --help`")
|
||||
}
|
||||
if input.IsSet(flagkey.FnRequestsPerPod) && isNotPoolManager {
|
||||
console.Warn("--requestsperpod is only valid for executortype; `poolmgr`. Check `fission function create --help`")
|
||||
}
|
||||
if input.IsSet(flagkey.FnOnceOnly) && isNotPoolManager {
|
||||
console.Warn("--onceonly is only valid for executortype; `poolmgr`. Check `fission function create --help`")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func getExecutorType(input cli.Input) (executorType fv1.ExecutorType, err error) {
|
||||
switch input.String(flagkey.FnExecutorType) {
|
||||
case "":
|
||||
|
||||
@@ -154,6 +154,11 @@ func (opts *UpdateSubCommand) complete(input cli.Input) error {
|
||||
function.Spec.IdleTimeout = &fnTimeout
|
||||
}
|
||||
|
||||
err = checkExecutorPoolManager(input, function.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if input.IsSet(flagkey.FnConcurrency) {
|
||||
function.Spec.Concurrency = input.Int(flagkey.FnConcurrency)
|
||||
}
|
||||
|
||||
@@ -71,7 +71,6 @@ func (opts *GetSubCommand) complete(input cli.Input) (err error) {
|
||||
}
|
||||
|
||||
func (opts *GetSubCommand) run(input cli.Input) error {
|
||||
|
||||
pkg, err := opts.Client().FissionClientSet.CoreV1().Packages(opts.namespace).Get(input.Context(), opts.name, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -79,7 +78,7 @@ func (opts *GetSubCommand) run(input cli.Input) error {
|
||||
|
||||
var reader io.Reader
|
||||
archive := pkg.Spec.Source
|
||||
if opts.archiveType == util.DEPLOY_ARCHIVE || archive.Type == "" {
|
||||
if (opts.archiveType == util.DEPLOY_ARCHIVE || archive.Type == "") && (pkg.Spec.Deployment.Type != "") {
|
||||
archive = pkg.Spec.Deployment
|
||||
}
|
||||
|
||||
|
||||
@@ -195,29 +195,21 @@ func DownloadStrorageURL(ctx context.Context, client cmd.Client, fileUrl string)
|
||||
}
|
||||
|
||||
func WriteArchiveToFile(fileName string, reader io.Reader) error {
|
||||
tmpDir, err := utils.GetTempDir()
|
||||
// Create the target file directly
|
||||
file, err := os.Create(fileName)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer file.Close()
|
||||
|
||||
// Copy data from the reader to the target file
|
||||
_, err = io.Copy(file, reader)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
tmpFileName := uuid.NewString()
|
||||
|
||||
path := filepath.Join(tmpDir, tmpFileName+".tmp")
|
||||
w, err := os.Create(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer w.Close()
|
||||
_, err = io.Copy(w, reader)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = os.Chmod(path, 0644)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = os.Rename(path, fileName)
|
||||
// Change the permissions of the target file
|
||||
err = os.Chmod(fileName, 0644)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -126,9 +126,9 @@ var (
|
||||
FnTestHeader = Flag{Type: StringSlice, Name: flagkey.FnTestHeader, Short: "H", Usage: "Request headers"}
|
||||
FnTestQuery = Flag{Type: StringSlice, Name: flagkey.FnTestQuery, Short: "q", Usage: "Request query parameters: -q key1=value1 -q key2=value2"}
|
||||
FnIdleTimeout = Flag{Type: Int, Name: flagkey.FnIdleTimeout, Usage: "The length of time (in seconds) that a function is idle before pod(s) are eligible for recycling", DefaultValue: 120}
|
||||
FnConcurrency = Flag{Type: Int, Name: flagkey.FnConcurrency, Aliases: []string{"con"}, Usage: "Maximum number of pods specialized concurrently to serve requests", DefaultValue: 500}
|
||||
FnRequestsPerPod = Flag{Type: Int, Name: flagkey.FnRequestsPerPod, Aliases: []string{"rpp"}, Usage: "Maximum number of concurrent requests that can be served by a specialized pod", DefaultValue: 1}
|
||||
FnOnceOnly = Flag{Type: Bool, Name: flagkey.FnOnceOnly, Aliases: []string{"yolo"}, Usage: "Specifies if specialized pod will serve exactly one request in its lifetime"}
|
||||
FnConcurrency = Flag{Type: Int, Name: flagkey.FnConcurrency, Aliases: []string{"con"}, Usage: "Maximum number of pods specialized concurrently to serve requests (Only valid for executortype; `poolmgr`)", DefaultValue: 500}
|
||||
FnRequestsPerPod = Flag{Type: Int, Name: flagkey.FnRequestsPerPod, Aliases: []string{"rpp"}, Usage: "Maximum number of concurrent requests that can be served by a specialized pod (Only valid for executortype; `poolmgr`)", DefaultValue: 1}
|
||||
FnOnceOnly = Flag{Type: Bool, Name: flagkey.FnOnceOnly, Aliases: []string{"yolo"}, Usage: "Specifies if specialized pod will serve exactly one request in its lifetime (Only valid for executortype; `poolmgr`)"}
|
||||
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."}
|
||||
FnRetainPods = Flag{Type: Int, Name: flagkey.FnRetainPods, Usage: "Number of pods to retain after pods specialization.", DefaultValue: 0}
|
||||
|
||||
@@ -74,6 +74,8 @@ func NewMqtConsumerGroupHandler(version sarama.KafkaVersion,
|
||||
|
||||
// Setup implemented to satisfy the sarama.ConsumerGroupHandler interface
|
||||
func (ch MqtConsumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error {
|
||||
mqtrigger.SetTriggerStatus(ch.trigger.ObjectMeta.Name, ch.trigger.ObjectMeta.Namespace)
|
||||
mqtrigger.IncreaseInprocessCount()
|
||||
ch.logger.With(
|
||||
zap.String("trigger", ch.trigger.ObjectMeta.Name),
|
||||
zap.String("topic", ch.trigger.Spec.Topic),
|
||||
@@ -88,6 +90,8 @@ func (ch MqtConsumerGroupHandler) Setup(session sarama.ConsumerGroupSession) err
|
||||
|
||||
// Cleanup implemented to satisfy the sarama.ConsumerGroupHandler interface
|
||||
func (ch MqtConsumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error {
|
||||
mqtrigger.ResetTriggerStatus(ch.trigger.ObjectMeta.Name, ch.trigger.ObjectMeta.Namespace)
|
||||
mqtrigger.DecreaseInprocessCount()
|
||||
ch.logger.With(
|
||||
zap.String("trigger", ch.trigger.ObjectMeta.Name),
|
||||
zap.String("topic", ch.trigger.Spec.Topic),
|
||||
|
||||
@@ -54,7 +54,6 @@ type (
|
||||
routerUrl string
|
||||
brokers []string
|
||||
version sarama.KafkaVersion
|
||||
client sarama.Client
|
||||
authKeys map[string][]byte
|
||||
tls bool
|
||||
}
|
||||
@@ -111,18 +110,24 @@ func New(logger *zap.Logger, mqCfg messageQueue.Config, routerUrl string) (messa
|
||||
|
||||
logger.Info("created kafka queue", zap.Any("kafka brokers", kafka.brokers),
|
||||
zap.Any("kafka version", kafka.version))
|
||||
return kafka, nil
|
||||
}
|
||||
|
||||
// Create new config
|
||||
saramaConfig := sarama.NewConfig()
|
||||
saramaConfig.Version = kafka.version
|
||||
func (kafka Kafka) Subscribe(trigger *fv1.MessageQueueTrigger) (messageQueue.Subscription, error) {
|
||||
kafka.logger.Debug("inside kakfa subscribe", zap.Any("trigger", trigger))
|
||||
kafka.logger.Debug("brokers set", zap.Strings("brokers", kafka.brokers))
|
||||
|
||||
// consumer config
|
||||
saramaConfig.Consumer.Return.Errors = true
|
||||
// Create new consumer
|
||||
consumerConfig := sarama.NewConfig()
|
||||
consumerConfig.Consumer.Return.Errors = true
|
||||
consumerConfig.Version = kafka.version
|
||||
|
||||
// producer config
|
||||
saramaConfig.Producer.RequiredAcks = sarama.WaitForAll
|
||||
saramaConfig.Producer.Retry.Max = 10
|
||||
saramaConfig.Producer.Return.Successes = true
|
||||
// Create new producer
|
||||
producerConfig := sarama.NewConfig()
|
||||
producerConfig.Producer.RequiredAcks = sarama.WaitForAll
|
||||
producerConfig.Producer.Retry.Max = 10
|
||||
producerConfig.Producer.Return.Successes = true
|
||||
producerConfig.Version = kafka.version
|
||||
|
||||
// Setup TLS for both producer and consumer
|
||||
if kafka.tls {
|
||||
@@ -132,30 +137,18 @@ func New(logger *zap.Logger, mqCfg messageQueue.Config, routerUrl string) (messa
|
||||
return nil, err
|
||||
}
|
||||
|
||||
saramaConfig.Net.TLS.Enable = true
|
||||
saramaConfig.Net.TLS.Config = tlsConfig
|
||||
producerConfig.Net.TLS.Enable = true
|
||||
producerConfig.Net.TLS.Config = tlsConfig
|
||||
consumerConfig.Net.TLS.Enable = true
|
||||
consumerConfig.Net.TLS.Config = tlsConfig
|
||||
}
|
||||
|
||||
saramaClient, err := sarama.NewClient(kafka.brokers, saramaConfig)
|
||||
consumer, err := sarama.NewConsumerGroup(kafka.brokers, string(trigger.ObjectMeta.UID), consumerConfig)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
kafka.client = saramaClient
|
||||
|
||||
return kafka, nil
|
||||
}
|
||||
|
||||
func (kafka Kafka) Subscribe(trigger *fv1.MessageQueueTrigger) (messageQueue.Subscription, error) {
|
||||
kafka.logger.Debug("inside kakfa subscribe", zap.Any("trigger", trigger))
|
||||
kafka.logger.Debug("brokers set", zap.Strings("brokers", kafka.brokers))
|
||||
|
||||
consumer, err := sarama.NewConsumerGroupFromClient(string(trigger.ObjectMeta.UID), kafka.client)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
producer, err := sarama.NewSyncProducerFromClient(kafka.client)
|
||||
producer, err := sarama.NewSyncProducer(kafka.brokers, producerConfig)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -45,6 +45,20 @@ var (
|
||||
},
|
||||
[]string{"trigger_name", "trigger_namespace", "topic", "partition"},
|
||||
)
|
||||
triggerStatus = prometheus.NewGaugeVec(
|
||||
prometheus.GaugeOpts{
|
||||
Name: "fission_mqt_status",
|
||||
Help: "Status of an individual trigger 1 if processing otherwise 0",
|
||||
},
|
||||
[]string{"trigger_name", "trigger_namespace"},
|
||||
)
|
||||
mqtInprocessCount = prometheus.NewGaugeVec(
|
||||
prometheus.GaugeOpts{
|
||||
Name: "fission_mqt_inprocess",
|
||||
Help: "Total number of MQTs in active processing",
|
||||
},
|
||||
[]string{},
|
||||
)
|
||||
)
|
||||
|
||||
func IncreaseSubscriptionCount() {
|
||||
@@ -55,6 +69,22 @@ func DecreaseSubscriptionCount() {
|
||||
subscriptionCount.WithLabelValues().Dec()
|
||||
}
|
||||
|
||||
func SetTriggerStatus(trigname, trignamespace string) {
|
||||
triggerStatus.WithLabelValues(trigname, trignamespace).Inc()
|
||||
}
|
||||
|
||||
func ResetTriggerStatus(trigname, trignamespace string) {
|
||||
triggerStatus.WithLabelValues(trigname, trignamespace).Dec()
|
||||
}
|
||||
|
||||
func IncreaseInprocessCount() {
|
||||
mqtInprocessCount.WithLabelValues().Inc()
|
||||
}
|
||||
|
||||
func DecreaseInprocessCount() {
|
||||
mqtInprocessCount.WithLabelValues().Dec()
|
||||
}
|
||||
|
||||
func IncreaseMessageCount(trigname, trignamespace string) {
|
||||
messageCount.WithLabelValues(trigname, trignamespace).Inc()
|
||||
}
|
||||
@@ -68,4 +98,6 @@ func init() {
|
||||
registry.MustRegister(subscriptionCount)
|
||||
registry.MustRegister(messageCount)
|
||||
registry.MustRegister(messageLagCount)
|
||||
registry.MustRegister(mqtInprocessCount)
|
||||
registry.MustRegister(triggerStatus)
|
||||
}
|
||||
|
||||
@@ -171,6 +171,10 @@ func IsURL(str string) bool {
|
||||
return strings.HasPrefix(str, "http://") || strings.HasPrefix(str, "https://")
|
||||
}
|
||||
|
||||
func isHttp2xxSuccessful(status int) bool {
|
||||
return status >= 200 && status < 300
|
||||
}
|
||||
|
||||
func DownloadUrl(ctx context.Context, httpClient *http.Client, url string, localPath string) error {
|
||||
resp, err := ctxhttp.Get(ctx, httpClient, url)
|
||||
if err != nil {
|
||||
@@ -178,6 +182,10 @@ func DownloadUrl(ctx context.Context, httpClient *http.Client, url string, local
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if !isHttp2xxSuccessful(resp.StatusCode) {
|
||||
return errors.New(resp.Status)
|
||||
}
|
||||
|
||||
w, err := os.Create(localPath)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -259,3 +267,32 @@ func FindFreePort() (int, error) {
|
||||
|
||||
return port, nil
|
||||
}
|
||||
|
||||
// DeleteOldPackages deletes src and built deployment packages from builder's storage.
|
||||
// The function also verifies that sharedVolumePath for builder and fetcher containers
|
||||
// is /packages. A source_package contains a directory and a .tmp file while a deployment
|
||||
// package contains a directory and a .zip file.
|
||||
func DeleteOldPackages(pkgPath, pkgType string) error {
|
||||
sharedVolumePath := "/packages"
|
||||
if !strings.HasPrefix(pkgPath, sharedVolumePath) {
|
||||
return fmt.Errorf("invalid shared volume path: %s", pkgPath)
|
||||
}
|
||||
|
||||
var file string
|
||||
if pkgType == "DEPLOY_PKG" {
|
||||
file = pkgPath + ".zip"
|
||||
} else if pkgType == "SRC_PKG" {
|
||||
file = pkgPath + ".tmp"
|
||||
}
|
||||
|
||||
err := os.RemoveAll(pkgPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = os.Remove(file)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user