Allow customization of logging configuration (#287)

Author: bnu0

api/v1beta1/flinkcluster_types.go

@@ -426,6 +426,11 @@ type FlinkClusterSpec struct {
 	// Config for GCP.
 	GCPConfig *GCPConfig `json:"gcpConfig,omitempty"`
 
+	// The logging configuration, which should have keys 'log4j-console.properties' and 'logback-console.xml'.
+	// These will end up in the 'flink-config-volume' ConfigMap, which gets mounted at /opt/flink/conf.
+	// If not provided, defaults that log to console only will be used.
+	LogConfig map[string]string `json:"logConfig,omitempty"`
+
 	// The maximum number of revision history to keep, default: 10.
 	RevisionHistoryLimit *int32 `json:"revisionHistoryLimit,omitempty"`
 }

api/v1beta1/zz_generated.deepcopy.go

@@ -3161,6 +3161,10 @@ spec:
               required:
               - accessScope
               type: object
+            logConfig:
+              additionalProperties:
+                type: string
+              type: object
             revisionHistoryLimit:
               format: int32
               type: integer

config/crd/bases/flinkoperator.k8s.io_flinkclusters.yaml

@@ -554,9 +554,9 @@ func getDesiredConfigMap(
 		}
 		flinkProps[k] = v
 	}
-	// TODO: Provide logging options: log4j-console.properties and log4j.properties
-	var log4jPropName = "log4j-console.properties"
-	var logbackXMLName = "logback-console.xml"
+	var configData = getLogConf(flinkCluster.Spec)
+	configData["flink-conf.yaml"] = getFlinkProperties(flinkProps)
+	configData["submit-job.sh"] = submitJobScript
 	var configMap = &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: clusterNamespace,
@@ -565,12 +565,7 @@ func getDesiredConfigMap(
 				ToOwnerReference(flinkCluster)},
 			Labels: labels,
 		},
-		Data: map[string]string{
-			"flink-conf.yaml": getFlinkProperties(flinkProps),
-			log4jPropName:     getLogConf()[log4jPropName],
-			logbackXMLName:    getLogConf()[logbackXMLName],
-			"submit-job.sh":   submitJobScript,
-		},
+		Data: configData,
 	}
 
 	return configMap
@@ -1051,10 +1046,8 @@ func mergeLabels(labels1 map[string]string, labels2 map[string]string) map[strin
 	return mergedLabels
 }
 
-// TODO: Wouldn't it be better to create a file, put it in an operator image, and read from them?.
-// Provide logging profiles
-func getLogConf() map[string]string {
-	var log4jConsoleProperties = `log4j.rootLogger=INFO, console
+const (
+	DefaultLog4jConfig = `log4j.rootLogger=INFO, console
 log4j.logger.akka=INFO
 log4j.logger.org.apache.kafka=INFO
 log4j.logger.org.apache.hadoop=INFO
@@ -1063,7 +1056,7 @@ log4j.appender.console=org.apache.log4j.ConsoleAppender
 log4j.appender.console.layout=org.apache.log4j.PatternLayout
 log4j.appender.console.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n
 log4j.logger.org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline=ERROR, console`
-	var logbackConsoleXML = `<configuration>
+	DefaultLogbackConfig = `<configuration>
     <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
         <encoder>
             <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{60} %X{sourceThread} - %msg%n</pattern>
@@ -1088,9 +1081,20 @@ log4j.logger.org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannel
         <appender-ref ref="console"/>
     </logger>
 </configuration>`
+)
 
-	return map[string]string{
-		"log4j-console.properties": log4jConsoleProperties,
-		"logback-console.xml":      logbackConsoleXML,
+// TODO: Wouldn't it be better to create a file, put it in an operator image, and read from them?.
+// Provide logging profiles
+func getLogConf(spec v1beta1.FlinkClusterSpec) map[string]string {
+	result := spec.LogConfig
+	if result == nil {
+		result = make(map[string]string, 2)
+	}
+	if _, isPresent := result["log4j-console.properties"]; !isPresent {
+		result["log4j-console.properties"] = DefaultLog4jConfig
+	}
+	if _, isPresent := result["logback-console.xml"]; !isPresent {
+		result["logback-console.xml"] = DefaultLogbackConfig
 	}
+	return result
 }

controllers/flinkcluster_converter.go

@@ -17,6 +17,7 @@ limitations under the License.
 package controllers
 
 import (
+	"reflect"
 	"testing"
 	"time"
 
@@ -254,6 +255,11 @@ func TestGetDesiredClusterState(t *testing.T) {
 					MountPath:  "/etc/gcp_service_account/",
 				},
 			},
+			LogConfig: map[string]string{
+				"extra-file.txt":           "hello!",
+				"log4j-console.properties": "foo",
+				"logback-console.xml":      "bar",
+			},
 		},
 		Status: v1beta1.FlinkClusterStatus{
 			NextRevision: "flinkjobcluster-sample-85dc8f749-1",
@@ -888,8 +894,9 @@ taskmanager.rpc.port: 6122
 		},
 		Data: map[string]string{
 			"flink-conf.yaml":          flinkConfYaml,
-			"log4j-console.properties": getLogConf()["log4j-console.properties"],
-			"logback-console.xml":      getLogConf()["logback-console.xml"],
+			"extra-file.txt":           "hello!",
+			"log4j-console.properties": "foo",
+			"logback-console.xml":      "bar",
 			"submit-job.sh":            submitJobScript,
 		},
 	}
@@ -969,3 +976,73 @@ func TestCalFlinkHeapSize(t *testing.T) {
 	flinkHeapSize = calFlinkHeapSize(cluster)
 	assert.Assert(t, len(flinkHeapSize) == 0)
 }
+
+func Test_getLogConf(t *testing.T) {
+	type args struct {
+		spec v1beta1.FlinkClusterSpec
+	}
+	tests := []struct {
+		name string
+		args args
+		want map[string]string
+	}{
+		{
+			name: "nil map uses defaults",
+			args: args{v1beta1.FlinkClusterSpec{LogConfig: nil}},
+			want: map[string]string{
+				"log4j-console.properties": DefaultLog4jConfig,
+				"logback-console.xml":      DefaultLogbackConfig,
+			},
+		},
+		{
+			name: "map missing log4j-console.properties uses default",
+			args: args{v1beta1.FlinkClusterSpec{LogConfig: map[string]string{
+				"logback-console.xml": "xyz",
+			}}},
+			want: map[string]string{
+				"log4j-console.properties": DefaultLog4jConfig,
+				"logback-console.xml":      "xyz",
+			},
+		},
+		{
+			name: "map missing logback-console.xml uses default",
+			args: args{v1beta1.FlinkClusterSpec{LogConfig: map[string]string{
+				"log4j-console.properties": "xyz",
+			}}},
+			want: map[string]string{
+				"log4j-console.properties": "xyz",
+				"logback-console.xml":      DefaultLogbackConfig,
+			},
+		},
+		{
+			name: "map with both keys overrides defaults",
+			args: args{v1beta1.FlinkClusterSpec{LogConfig: map[string]string{
+				"log4j-console.properties": "hello",
+				"logback-console.xml":      "world",
+			}}},
+			want: map[string]string{
+				"log4j-console.properties": "hello",
+				"logback-console.xml":      "world",
+			},
+		},
+		{
+			name: "extra keys preserved",
+			args: args{v1beta1.FlinkClusterSpec{LogConfig: map[string]string{
+				"log4j-console.properties": "abc",
+				"file.txt":      						"def",
+			}}},
+			want: map[string]string{
+				"log4j-console.properties": "abc",
+				"logback-console.xml": 			DefaultLogbackConfig,
+				"file.txt":      						"def",
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := getLogConf(tt.args.spec); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("getLogConf() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

controllers/flinkcluster_converter_test.go

@@ -88,6 +88,7 @@ FlinkCluster
             |__ secretName
             |__ keyFile
             |__ mountPath
+    |__ logConfig
     |__ revisionHistoryLimit
 |__ status
     |__ state
@@ -279,6 +280,13 @@ FlinkCluster
           same namespace as the FlinkCluster.
         * **keyFile**: The name of the service account key file.
         * **mountPath**: The path where to mount the Volume of the Secret.
+    * **logConfig** (optional): The logging configuration, a string-to-string map that becomes the ConfigMap mounted at 
+      `/opt/flink/conf` in launched Flink pods. See below for some possible keys to include:
+      * **log4j-console.properties**: The contents of the log4j properties file to use. If not provided, a default that 
+        logs only to stdout will be provided.
+      * **logback-console.xml**: The contents of the logback XML file to use. If not provided, a default that logs only 
+        to stdout will be provided.
+      * Other arbitrary keys are also allowed, and will become part of the ConfigMap.
     * **revisionHistoryLimit** (optional): The maximum number of revision history to keep, default: 10.
   * **status**: Flink job or session cluster status.
     * **state**: The overall state of the Flink cluster.

docs/crd.md

@@ -444,4 +444,23 @@ You can check the list of revisions and their contents like this:
 ```bash
 kubectl get controllerrevision
 kubectl get controllerrevision <REVISION-NAME> -o yaml
-```
+```
+
+### Control Logging Behavior
+
+The default logging configuration provided by the operator sends logs from JobManager and TaskManager to `stdout`. This
+has the effect of making it so that logging from Flink workloads running on Kubernetes behaves like every other 
+Kubernetes pod. Your Flink logs should be stored wherever you generally expect to see your container logs in your 
+environment.
+
+Sometimes, however, this is not a good fit. An example of when you might want to customize logging behavior is to 
+restore the visibility of logs in the Flink JobManager web interface. Or you might want to ship logs directly to a 
+different sink, or using a different formatter. 
+
+You can use the `spec.logConfig` field to fully control the log4j and logback configuration. It is a string-to-string map,
+whose keys and values become filenames and contents (respectively) in the folder `/opt/flink/conf` in each container. 
+The default Flink docker entrypoint expects this directory to contain two files: `log4j-console.properties` and 
+`logback-console.xml`.
+
+An example of using this parameter to make logs visible in both the Flink UI and on stdout 
+[can be found here](../examples/log_config.yaml).

docs/user_guide.md

@@ -0,0 +1,69 @@
+apiVersion: flinkoperator.k8s.io/v1beta1
+kind: FlinkCluster
+metadata:
+  name: my-cool-cluster
+spec:
+  image:
+    name: flink:1.11.1
+  jobManager:
+    resources:
+      limits:
+        memory: 1Gi
+        cpu: "1.0"
+  taskManager:
+    replicas: 2
+    resources:
+      limits:
+        memory: 1Gi
+        cpu: "1.0"
+  flinkProperties:
+    taskmanager.numberOfTaskSlots: "1"
+  logConfig:
+    "log4j-console.properties": |
+      rootLogger.level = INFO
+      rootLogger.appenderRef.file.ref = LogFile
+      rootLogger.appenderRef.console.ref = LogConsole
+      appender.file.name = LogFile
+      appender.file.type = File
+      appender.file.append = false
+      appender.file.fileName = ${sys:log.file}
+      appender.file.layout.type = PatternLayout
+      appender.file.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n
+      appender.console.name = LogConsole
+      appender.console.type = CONSOLE
+      appender.console.layout.type = PatternLayout
+      appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n
+      logger.akka.name = akka
+      logger.akka.level = INFO
+      logger.kafka.name= org.apache.kafka
+      logger.kafka.level = INFO
+      logger.hadoop.name = org.apache.hadoop
+      logger.hadoop.level = INFO
+      logger.zookeeper.name = org.apache.zookeeper
+      logger.zookeeper.level = INFO
+      logger.netty.name = org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline
+      logger.netty.level = OFF
+    "logback-console.xml": |
+      <configuration>
+        <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
+          <encoder>
+            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{60} %X{sourceThread} - %msg%n</pattern>
+          </encoder>
+        </appender>
+        <appender name="file" class="ch.qos.logback.core.FileAppender">
+          <file>${log.file}</file>
+          <append>false</append>
+          <encoder>
+            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{60} %X{sourceThread} - %msg%n</pattern>
+          </encoder>
+        </appender>
+        <root level="INFO">
+          <appender-ref ref="console"/>
+          <appender-ref ref="file"/>
+        </root>
+        <logger name="akka" level="INFO" />
+        <logger name="org.apache.kafka" level="INFO" />
+        <logger name="org.apache.hadoop" level="INFO" />
+        <logger name="org.apache.zookeeper" level="INFO" />
+        <logger name="org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline" level="ERROR" />
+      </configuration>