fix(fluentd): use Fluentd's default ports for inputs without a port

pkg/operator/fluentd-service.go

@@ -56,9 +56,9 @@ func MakeFluentdService(fd fluentdv1alpha1.Fluentd) *corev1.Service {
 	for _, input := range globalInputs {
 
 		if input.Forward != nil {
-			forwardPort := *input.Forward.Port
-			if forwardPort == 0 {
-				forwardPort = DefaultForwardPort
+			forwardPort := DefaultForwardPort
+			if input.Forward.Port != nil {
+				forwardPort = *input.Forward.Port
 			}
 
 			forwardContainerPort := corev1.ServicePort{
@@ -72,9 +72,9 @@ func MakeFluentdService(fd fluentdv1alpha1.Fluentd) *corev1.Service {
 		}
 
 		if input.Http != nil {
-			httpPort := *input.Http.Port
-			if httpPort == 0 {
-				httpPort = DefaultHttpPort
+			httpPort := DefaultHttpPort
+			if input.Http.Port != nil {
+				httpPort = *input.Http.Port
 			}
 			httpContainerPort := corev1.ServicePort{
 				Name:       DefaultHttpName,

pkg/operator/fluentd_service_test.go

@@ -0,0 +1,73 @@
+package operator
+
+import (
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+
+	fluentdv1alpha1 "github.com/fluent/fluent-operator/v3/apis/fluentd/v1alpha1"
+	"github.com/fluent/fluent-operator/v3/apis/fluentd/v1alpha1/plugins/input"
+)
+
+// Inputs that leave their port unset listen on Fluentd's defaults, so the
+// container ports and the Service must point there too.
+func TestFluentdInputPorts(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		input    input.Input
+		portName string
+		want     int32
+	}{
+		{
+			name:     "forward without a port",
+			input:    input.Input{Forward: &input.Forward{}},
+			portName: DefaultForwardName,
+			want:     24224,
+		},
+		{
+			name:     "forward with a port",
+			input:    input.Input{Forward: &input.Forward{Port: new(int32(24225))}},
+			portName: DefaultForwardName,
+			want:     24225,
+		},
+		{
+			name:     "http without a port",
+			input:    input.Input{Http: &input.Http{}},
+			portName: DefaultHttpName,
+			want:     9880,
+		},
+		{
+			name:     "http with a port",
+			input:    input.Input{Http: &input.Http{Port: new(int32(9881))}},
+			portName: DefaultHttpName,
+			want:     9881,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			fd := fluentdv1alpha1.Fluentd{
+				ObjectMeta: metav1.ObjectMeta{Name: "fluentd", Namespace: "logging"},
+				Spec:       fluentdv1alpha1.FluentdSpec{GlobalInputs: []input.Input{tc.input}},
+			}
+
+			var containerPort *corev1.ContainerPort
+			for _, p := range makeFluentdPorts(fd) {
+				if p.Name == tc.portName {
+					containerPort = &p
+				}
+			}
+			if containerPort == nil || containerPort.ContainerPort != tc.want {
+				t.Fatalf("container port %q = %v, want %d", tc.portName, containerPort, tc.want)
+			}
+
+			svc := MakeFluentdService(fd)
+			if len(svc.Spec.Ports) != 1 {
+				t.Fatalf("service ports = %v, want one", svc.Spec.Ports)
+			}
+			if got := svc.Spec.Ports[0]; got.Port != tc.want || got.TargetPort != intstr.FromString(tc.portName) {
+				t.Fatalf("service port = %+v, want port %d targeting %q", got, tc.want, tc.portName)
+			}
+		})
+	}
+}

pkg/operator/sts.go

@@ -19,7 +19,8 @@ const (
 
 	MetricsPort int32 = 2021
 
-	DefaultForwardPort int32 = 24424
+	// Fluentd's own defaults, used when an input leaves its port unset.
+	DefaultForwardPort int32 = 24224
 	DefaultHttpPort    int32 = 9880
 	// 101 is the fsGroup that fluentd runs as in the kubesphere image
 	DefaultFsGroup int64 = 101