Invoking k8s service with proper k8s domain

pkg/event-consumers/kafka/kafka-consumer.go

@@ -28,12 +28,13 @@ import (
 )
 
 var (
-	stopM     map[string](chan struct{})
-	stoppedM  map[string](chan struct{})
-	consumerM map[string]bool
-	brokers   string
-	logLevel  string
-	config    *cluster.Config
+	stopM         map[string](chan struct{})
+	stoppedM      map[string](chan struct{})
+	consumerM     map[string]bool
+	brokers       string
+	logLevel      string
+	config        *cluster.Config
+	clusterDomain string
 )
 
 func init() {
@@ -80,6 +81,9 @@ func init() {
 		}
 	}
 
+	// local utils
+	clusterDomain = utils.GetClusterDomain()
+
 }
 
 // createConsumerProcess gets messages to a Kafka topic from the broker and send the payload to function service
@@ -106,7 +110,7 @@ func createConsumerProcess(broker, topic, funcName, ns, consumerGroupID string,
 				logrus.Debugf("Sending message %v to function %s", msg, funcName)
 				consumer.MarkOffset(msg, "")
 				go func() {
-					req, err := utils.GetHTTPReq(clientset, funcName, ns, "kafkatriggers.kubeless.io", "POST", string(msg.Value))
+					req, err := utils.GetHTTPReq(clientset, funcName, ns, clusterDomain, "kafkatriggers.kubeless.io", "POST", string(msg.Value))
 					if err != nil {
 						logrus.Errorf("Unable to elaborate request: %v", err)
 					} else {

pkg/utils/event_sender.go

@@ -46,14 +46,14 @@ func GetFunctionPort(clientset kubernetes.Interface, namespace, functionName str
 }
 
 // GetHTTPReq returns the http request object that can be used to send a event with payload to function service
-func GetHTTPReq(clientset kubernetes.Interface, funcName, namespace, eventNamespace, method, body string) (*http.Request, error) {
+func GetHTTPReq(clientset kubernetes.Interface, funcName, namespace, clusterDomain, eventNamespace, method, body string) (*http.Request, error) {
 
 	funcPort, err := GetFunctionPort(clientset, namespace, funcName)
 	if err != nil {
 		return nil, err
 	}
 
-	req, err := http.NewRequest(method, fmt.Sprintf("http://%s.%s.svc.cluster.local:%s", funcName, namespace, funcPort), strings.NewReader(body))
+	req, err := http.NewRequest(method, fmt.Sprintf("http://%s.%s.svc.%s:%s", funcName, namespace, clusterDomain, funcPort), strings.NewReader(body))
 	if err != nil {
 		return nil, fmt.Errorf("Unable to create request %v", err)
 	}

pkg/utils/event_sender_test.go

@@ -40,7 +40,7 @@ func TestGetHTTPRequest(t *testing.T) {
 		},
 	}
 	clientset := fake.NewSimpleClientset(&svc)
-	req, err := GetHTTPReq(clientset, "foo", "myns", "kafkatriggers.kubeless.io", "POST", "my msg")
+	req, err := GetHTTPReq(clientset, "foo", "myns", "cluster.local", "kafkatriggers.kubeless.io", "POST", "my msg")
 	if err != nil {
 		t.Errorf("Unexpected error %v", err)
 	}
@@ -90,7 +90,7 @@ func TestGetJSONHTTPRequest(t *testing.T) {
 		},
 	}
 	clientset := fake.NewSimpleClientset(&svc)
-	req, err := GetHTTPReq(clientset, "foo", "myns", "kafkatriggers.kubeless.io", "POST", `{"hello": "world"}`)
+	req, err := GetHTTPReq(clientset, "foo", "myns", "cluster.local", "kafkatriggers.kubeless.io", "POST", `{"hello": "world"}`)
 	if err != nil {
 		t.Errorf("Unexpected error %v", err)
 	}

pkg/utils/k8sutil.go

@@ -21,8 +21,10 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"net"
 	"os"
 	"path/filepath"
+	"strings"
 
 	kafkaApi "github.com/kubeless/kafka-trigger/pkg/apis/kubeless/v1beta1"
 	"github.com/sirupsen/logrus"
@@ -312,3 +314,20 @@ func GetSecretsAsLocalObjectReference(secrets ...string) []v1.LocalObjectReferen
 	}
 	return res
 }
+
+// GetClusterDomain returns Kubernetes cluster domain, default to "cluster.local"
+func GetClusterDomain() string {
+	apiSvc := "kubernetes.default.svc."
+
+	clusterDomain := "cluster.local"
+
+	cname, err := net.LookupCNAME(apiSvc)
+	if err != nil {
+		return clusterDomain
+	}
+
+	clusterDomain = strings.TrimPrefix(cname, apiSvc)
+	clusterDomain = strings.TrimSuffix(clusterDomain, ".")
+
+	return clusterDomain
+}