Use wget for http/https package in downloader container

.ci/clusters/nginx.yaml

@@ -0,0 +1,66 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: nginx-demo
+data:
+  default.conf: |
+    server {
+      listen       80;
+
+      # download
+      autoindex on;               # enable directory listing output
+      autoindex_exact_size off;   # output file sizes rounded to kilobytes, megabytes, and gigabytes
+      autoindex_localtime on;     # output local times in the directory
+
+      location / {
+        root /tmp;
+      }
+    }
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: nginx-server
+  labels:
+    app: nginx
+spec:
+  ports:
+    - name: http
+      port: 80
+      targetPort: 80
+      protocol: TCP
+  type: LoadBalancer
+  selector:
+    app: nginx
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: nginx
+spec:
+  selector:
+    matchLabels:
+      app: nginx
+  serviceName: nginx-server
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        app: nginx
+    spec:
+      containers:
+        - name: nginx
+          image: nginx:1.14.2
+          ports:
+            - containerPort: 80
+          volumeMounts:
+            - name: nginx-config
+              mountPath: /etc/nginx/conf.d/default.conf
+              subPath: default.conf
+      volumes:
+        - name: nginx-config
+          configMap:
+            name: nginx-demo
+            items:
+              - key: default.conf
+                path: default.conf

.ci/helm.sh

@@ -283,6 +283,10 @@ function ci::verify_download_python_legacy_function_oauth2() {
     fi
 }
 
+function ci::verify_download_from_http_python_function() {
+    ci::verify_exclamation_function_with_auth "persistent://public/default/input-download-from-http-python-topic" "persistent://public/default/output-download-from-http-python-topic" "test-message" "test-message!" 10
+}
+
 function ci::verify_download_python_zip_function() {
     ci::verify_exclamation_function "persistent://public/default/input-download-python-zip-topic" "persistent://public/default/output-download-python-zip-topic" "test-message" "test-message!" 10
 }

.ci/tests/integration-oauth2/cases/py-download-from-http-function/manifests.yaml

@@ -0,0 +1,67 @@
+apiVersion: compute.functionmesh.io/v1alpha1
+kind: Function
+metadata:
+  name: py-function-download-from-http-sample
+  namespace: default
+spec:
+  className: exclamation
+  forwardSourceMessageProperty: true
+  maxPendingAsyncRequests: 1000
+  replicas: 1
+  maxReplicas: 1
+  logTopic: persistent://public/default/py-function-logs
+  input:
+    topics:
+      - persistent://public/default/input-download-from-http-python-topic
+  output:
+    topic: persistent://public/default/output-download-from-http-python-topic
+  resources:
+    requests:
+      cpu: 50m
+      memory: 1G
+    limits:
+      cpu: "0.2"
+      memory: 1.1G
+  # each secret will be loaded ad an env variable from the `path` secret with the `key` in that secret in the name of `name`
+  secretsMap:
+    "name":
+      path: "test-py-secret"
+      key: "username"
+    "pwd":
+      path: "test-py-secret"
+      key: "password"
+  pulsar:
+    pulsarConfig: "test-py-pulsar"
+    tlsConfig:
+      enabled: false
+      allowInsecure: true
+      hostnameVerification: true
+    authConfig:
+      oauth2Config:
+        audience: api://56c1bd14-3ba7-4804-b47b-d46de6dce33e/.default
+        issuerUrl: https://sts.windows.net/06a8a086-ae6e-45b5-a22e-ad90de23013e/v2.0
+        scope: api://56c1bd14-3ba7-4804-b47b-d46de6dce33e/.default
+        keySecretName: sn-platform-oauth2-private-key
+        keySecretKey: auth.json
+  python:
+    py: /pulsar/exclamation.zip
+    pyLocation: http://nginx-server.default.svc.cluster.local/exclamation.zip
+  clusterName: test
+  autoAck: true
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-py-pulsar
+data:
+  webServiceURL: http://sn-platform-pulsar-broker.default.svc.cluster.local:8080
+  brokerServiceURL: pulsar://sn-platform-pulsar-broker.default.svc.cluster.local:6650
+---
+apiVersion: v1
+data:
+  username: YWRtaW4=
+  password: MWYyZDFlMmU2N2Rm
+kind: Secret
+metadata:
+  name: test-py-secret
+type: Opaque

.ci/tests/integration-oauth2/cases/py-download-from-http-function/verify.sh

@@ -0,0 +1,50 @@
+#!/usr/bin/env bash
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+#
+
+set -e
+
+E2E_DIR=$(dirname "$0")
+BASE_DIR=$(cd "${E2E_DIR}"/../../../../..;pwd)
+PULSAR_NAMESPACE=${PULSAR_NAMESPACE:-"default"}
+PULSAR_RELEASE_NAME=${PULSAR_RELEASE_NAME:-"sn-platform"}
+E2E_KUBECONFIG=${E2E_KUBECONFIG:-"/tmp/e2e-k8s.config"}
+
+source "${BASE_DIR}"/.ci/helm.sh
+
+if [ ! "$KUBECONFIG" ]; then
+  export KUBECONFIG=${E2E_KUBECONFIG}
+fi
+
+kubectl apply -f "${BASE_DIR}"/.ci/tests/integration-oauth2/cases/py-download-from-http-function/manifests.yaml > /dev/null 2>&1
+
+verify_fm_result=$(ci::verify_function_mesh py-function-download-from-http-sample 2>&1)
+if [ $? -ne 0 ]; then
+  echo "$verify_fm_result"
+  kubectl delete -f "${BASE_DIR}"/.ci/tests/integration-oauth2/cases/py-download-from-http-function/manifests.yaml > /dev/null 2>&1 || true
+  exit 1
+fi
+
+verify_python_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::verify_download_from_http_python_function true 2>&1)
+if [ $? -eq 0 ]; then
+  echo "e2e-test: ok" | yq eval -
+else
+  echo "$verify_python_result"
+fi
+kubectl delete -f "${BASE_DIR}"/.ci/tests/integration-oauth2/cases/py-download-from-http-function/manifests.yaml > /dev/null 2>&1 || true

.ci/tests/integration-oauth2/e2e_with_downloader.yaml

@@ -72,6 +72,20 @@ setup:
         bash .ci/upload_function_with_oauth.sh java
         bash .ci/upload_function_with_oauth.sh py
 
+    # testing download packages from http
+    - name: start nginx http server
+      command: |
+        kubectl apply -f .ci/clusters/nginx.yaml
+      wait:
+        - namespace: default
+          resource: pod
+          label-selector: app=nginx
+          for: condition=Ready
+
+    - name: upload pyzip package
+      command: |
+        kubectl cp .ci/examples/py-examples/exclamation.zip nginx-0:/tmp
+
     - name: install function-mesh operator
       command: |
         make generate
@@ -107,3 +121,5 @@ verify:
       expected: expected.data.yaml
     - query: timeout 5m bash .ci/tests/integration-oauth2/cases/py-download-function-legacy/verify.sh
       expected: expected.data.yaml
+    - query: timeout 5m bash .ci/tests/integration-oauth2/cases/py-download-from-http-function/verify.sh
+      expected: expected.data.yaml

controllers/spec/common.go

@@ -75,6 +75,9 @@ const (
 	PackageNameSinkPrefix     = "sink://"
 	PackageNameSourcePrefix   = "source://"
 
+	HTTPPrefix  = "http://"
+	HTTPSPrefix = "https://"
+
 	AnnotationPrometheusScrape = "prometheus.io/scrape"
 	AnnotationPrometheusPort   = "prometheus.io/port"
 	AnnotationManaged          = "compute.functionmesh.io/managed"
@@ -248,16 +251,6 @@ func MakeStatefulSet(objectMeta *metav1.ObjectMeta, replicas *int32, downloaderI
 	var podVolumes = volumes
 	// there must be a download path specified, we need to create an init container and emptyDir volume
 	if len(volumeMounts) > 0 {
-		if pulsar.AuthConfig != nil && pulsar.AuthConfig.OAuth2Config != nil {
-			volumeMounts = append(volumeMounts, generateVolumeMountFromOAuth2Config(pulsar.AuthConfig.OAuth2Config))
-		}
-
-		if !reflect.ValueOf(pulsar.TLSConfig).IsNil() && pulsar.TLSConfig.HasSecretVolume() {
-			volumeMounts = append(volumeMounts, generateVolumeMountFromTLSConfig(pulsar.TLSConfig))
-		}
-
-		volumeMounts = append(volumeMounts, definedVolumeMounts...)
-
 		var downloadPath, componentPackage string
 		if javaRuntime != nil {
 			downloadPath = javaRuntime.JarLocation
@@ -270,6 +263,18 @@ func MakeStatefulSet(objectMeta *metav1.ObjectMeta, replicas *int32, downloaderI
 			componentPackage = goRuntime.Go
 		}
 
+		// mount auth and tls related VolumeMounts when download package from pulsar
+		if !hasHTTPPrefix(downloadPath) {
+			if pulsar.AuthConfig != nil && pulsar.AuthConfig.OAuth2Config != nil {
+				volumeMounts = append(volumeMounts, generateVolumeMountFromOAuth2Config(pulsar.AuthConfig.OAuth2Config))
+			}
+
+			if !reflect.ValueOf(pulsar.TLSConfig).IsNil() && pulsar.TLSConfig.HasSecretVolume() {
+				volumeMounts = append(volumeMounts, generateVolumeMountFromTLSConfig(pulsar.TLSConfig))
+			}
+		}
+		volumeMounts = append(volumeMounts, definedVolumeMounts...)
+
 		image := downloaderImage
 		if image == "" {
 			image = DownloaderImage
@@ -511,6 +516,10 @@ func getLegacyDownloadCommand(downloadPath, componentPackage string, authProvide
 func getDownloadCommand(downloadPath, componentPackage string, tlsProvided, authProvided bool, tlsConfig TLSConfig,
 	authConfig *v1alpha1.AuthConfig) []string {
 	var args []string
+	if hasHTTPPrefix(downloadPath) {
+		args = append(args, "wget", downloadPath, "-O", componentPackage)
+		return args
+	}
 	// activate oauth2 for pulsarctl
 	if authConfig != nil && authConfig.OAuth2Config != nil {
 		args = []string{
@@ -833,6 +842,11 @@ func hasPackageNamePrefix(packagesName string) bool {
 		strings.HasPrefix(packagesName, PackageNameSourcePrefix)
 }
 
+func hasHTTPPrefix(packageName string) bool {
+	return strings.HasPrefix(packageName, HTTPPrefix) ||
+		strings.HasPrefix(packageName, HTTPSPrefix)
+}
+
 func setShardIDEnvironmentVariableCommand() string {
 	return fmt.Sprintf("%s=${POD_NAME##*-} && echo shardId=${%s}", EnvShardID, EnvShardID)
 }

controllers/spec/common_test.go

@@ -164,6 +164,23 @@ func TestGetDownloadCommand(t *testing.T) {
 				"packages", "download", "function://public/default/test@v1", "--path", "function-package.jar",
 			},
 		},
+		{"http://aaa.bbb.ccc/test.jar", "function-package.jar",
+			&v1alpha1.PulsarTLSConfig{
+				TLSConfig: v1alpha1.TLSConfig{
+					Enabled:              true,
+					AllowInsecure:        true,
+					HostnameVerification: false,
+					CertSecretName:       "test-secret",
+					CertSecretKey:        "test-key",
+				},
+			}, nil,
+			[]string{
+				"wget",
+				"http://aaa.bbb.ccc/test.jar",
+				"-O",
+				"function-package.jar",
+			},
+		},
 	}
 
 	for _, v := range testData {