diff --git a/controllers/static/zookeeper/fluent-bit.conf.tmpl b/controllers/static/zookeeper/fluent-bit.conf.tmpl
new file mode 100644
index 00000000..c3c32885
--- /dev/null
+++ b/controllers/static/zookeeper/fluent-bit.conf.tmpl
@@ -0,0 +1,35 @@
+[SERVICE]
+    http_server On
+    http_port   2020
+    log_level   debug
+[INPUT]
+    name                tail
+    tag                 ${K8S_NAMESPACE}.${K8S_NODENAME}.${K8S_PODNAME}.zookeeper
+    path                /watch/*.log
+    path_key            full_path
+    refresh_interval    5
+    read_from_head      True
+    db                  /watch/zk_fluentbit.db
+[FILTER]
+    name    modify
+    match   *
+    add     namespace ${K8S_NAMESPACE}
+    add     nodeName ${K8S_NODENAME}
+    add     podName ${K8S_PODNAME}
+    add     ip ${K8S_PODIP}
+    add     labels_run zookeeper
+    add     component zookeeper
+{{- range .ExtraKeys }}
+    add     {{ .Key }} ${K8S_{{ .Value -}}}
+{{- end }}
+[OUTPUT]
+    name    stdout
+    match   *
+    format  json_lines
+[OUTPUT]
+    name    http
+    match   *
+    uri     /${K8S_NAMESPACE}.${K8S_NODENAME}.${K8S_PODNAME}.dib
+    format  json
+    host    {{ .FluentBitHTTPInputHost }}
+    port    {{ .FluentBitHTTPInputPort }}
diff --git a/controllers/static/zookeeper/logback.xml b/controllers/static/zookeeper/logback.xml
new file mode 100644
index 00000000..40eb6a10
--- /dev/null
+++ b/controllers/static/zookeeper/logback.xml
@@ -0,0 +1,119 @@
+<!--
+ Copyright 2022 The Apache Software Foundation
+
+ 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.
+
+ Define some default values that can be overridden by system properties
+-->
+<configuration>
+    <!-- Uncomment this if you would like to expose Logback JMX beans -->
+    <!--jmxConfigurator
+    /-->
+
+    <property name="zookeeper.console.threshold" value="INFO" />
+
+    <property name="zookeeper.log.dir" value="/var/log/" />
+    <property name="zookeeper.log.file" value="zookeeper.log" />
+    <property name="zookeeper.log.threshold" value="INFO" />
+    <property name="zookeeper.log.maxfilesize" value="256MB" />
+    <property name="zookeeper.log.maxbackupindex" value="2" />
+
+    <!--
+    console
+    Add "console" to root logger if you want to use this
+  -->
+    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L] - %m%n</pattern>
+        </encoder>
+        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
+            <level>${zookeeper.console.threshold}</level>
+        </filter>
+    </appender>
+
+    <!--
+    Add ROLLINGFILE to root logger to get log file output
+  -->
+    <appender name="ROLLINGFILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <File>${zookeeper.log.dir}/${zookeeper.log.file}</File>
+        <encoder>
+            <pattern>%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L] - %m%n</pattern>
+        </encoder>
+        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
+            <level>${zookeeper.log.threshold}</level>
+        </filter>
+        <rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
+            <maxIndex>${zookeeper.log.maxbackupindex}</maxIndex>
+            <FileNamePattern>${zookeeper.log.dir}/${zookeeper.log.file}.%i</FileNamePattern>
+        </rollingPolicy>
+        <triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
+            <MaxFileSize>${zookeeper.log.maxfilesize}</MaxFileSize>
+        </triggeringPolicy>
+    </appender>
+
+    <!--
+    Add TRACEFILE to root logger to get log file output
+    Log TRACE level and above messages to a log file
+  -->
+    <!--property
+    name="zookeeper.tracelog.dir" value="${zookeeper.log.dir}" />
+  <property name="zookeeper.tracelog.file" value="zookeeper_trace.log" />
+  <appender name="TRACEFILE" class="ch.qos.logback.core.FileAppender">
+    <File>${zookeeper.tracelog.dir}/${zookeeper.tracelog.file}</File>
+    <encoder>
+      <pattern>%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L] - %m%n</pattern>
+    </encoder>
+    <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
+      <level>TRACE</level>
+    </filter>
+  </appender-->
+
+    <!--
+    zk audit logging
+  -->
+    <!--property
+    name="zookeeper.auditlog.file" value="zookeeper_audit.log" />
+  <property name="zookeeper.auditlog.threshold" value="INFO" />
+  <property name="audit.logger" value="INFO, RFAAUDIT" />
+
+  <appender name="RFAAUDIT" class="ch.qos.logback.core.rolling.RollingFileAppender">
+    <File>${zookeeper.log.dir}/${zookeeper.auditlog.file}</File>
+    <encoder>
+      <pattern>%d{ISO8601} %p %c{2}: %m%n</pattern>
+    </encoder>
+    <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
+      <level>${zookeeper.auditlog.threshold}</level>
+    </filter>
+    <rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
+      <maxIndex>10</maxIndex>
+      <FileNamePattern>${zookeeper.log.dir}/${zookeeper.auditlog.file}.%i</FileNamePattern>
+    </rollingPolicy>
+    <triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
+      <MaxFileSize>10MB</MaxFileSize>
+    </triggeringPolicy>
+  </appender>
+
+  <logger name="org.apache.zookeeper.audit.Slf4jAuditLogger" additivity="false"
+    level="${audit.logger}">
+    <appender-ref ref="RFAAUDIT" />
+  </logger-->
+
+    <root level="INFO">
+        <appender-ref ref="CONSOLE" />
+        <appender-ref ref="ROLLINGFILE" />
+    </root>
+</configuration>
\ No newline at end of file
diff --git a/controllers/static/zookeeper/run.sh b/controllers/static/zookeeper/run.sh
index dc84bad4..d4fd9ce1 100755
--- a/controllers/static/zookeeper/run.sh
+++ b/controllers/static/zookeeper/run.sh
@@ -98,7 +98,7 @@ do
     echo "server.$i=$NAME-$((i-1)).$DOMAIN:$ZK_SERVER_PORT:$ZK_ELECTION_PORT" >> "$ZK_CONFIG_FILE"
 done
 
-cp /zookeeper/conf/logback.xml /conf
+cp /config-scripts/logback.xml /conf
 
 if [ -n "$JMXDISABLE" ]
 then
diff --git a/controllers/zookeeper.go b/controllers/zookeeper.go
index bf0252a6..419549f2 100644
--- a/controllers/zookeeper.go
+++ b/controllers/zookeeper.go
@@ -5,11 +5,13 @@ package controllers
 
 import (
 	_ "embed"
+	"strconv"
 
 	certv1 "github.com/cert-manager/cert-manager/pkg/apis/certmanager/v1"
 	"github.com/softwarefactory-project/sf-operator/controllers/libs/base"
 	"github.com/softwarefactory-project/sf-operator/controllers/libs/cert"
 	"github.com/softwarefactory-project/sf-operator/controllers/libs/conds"
+	logging "github.com/softwarefactory-project/sf-operator/controllers/libs/logging"
 	"github.com/softwarefactory-project/sf-operator/controllers/libs/utils"
 	appsv1 "k8s.io/api/apps/v1"
 	apiv1 "k8s.io/api/core/v1"
@@ -22,7 +24,13 @@ var zookeeperOk string
 var zookeeperReady string
 
 //go:embed static/zookeeper/run.sh
-var zookeepeRun string
+var zookeeperRun string
+
+//go:embed static/zookeeper/fluent-bit.conf.tmpl
+var zkFluentBitForwarderConfig string
+
+//go:embed static/zookeeper/logback.xml
+var zkLogbackConfig string
 
 const zkPortName = "zk"
 const zkPort = 2181
@@ -39,6 +47,36 @@ const zkServerPort = 2888
 const zkIdent = "zookeeper"
 const zkPIMountPath = "/config-scripts"
 
+func createZKLogForwarderSidecar(r *SFController, annotations map[string]string) (apiv1.Volume, apiv1.Container) {
+
+	fbForwarderConfig := make(map[string]string)
+	fbForwarderConfig["fluent-bit.conf"], _ = utils.ParseString(
+		zkFluentBitForwarderConfig,
+		struct {
+			ExtraKeys              []logging.FluentBitLabel
+			FluentBitHTTPInputHost string
+			FluentBitHTTPInputPort string
+		}{[]logging.FluentBitLabel{}, r.cr.Spec.FluentBitLogForwarding.HTTPInputHost, strconv.Itoa(int(r.cr.Spec.FluentBitLogForwarding.HTTPInputPort))})
+	r.EnsureConfigMap("fluentbit-zk-cfg", fbForwarderConfig)
+
+	volume := base.MkVolumeCM("zk-log-forwarder-config",
+		"fluentbit-zk-cfg-config-map")
+
+	volumeMounts := []apiv1.VolumeMount{
+		{
+			Name:      zkIdent + "-logs",
+			MountPath: "/watch/",
+		},
+		{
+			Name:      "zk-log-forwarder-config",
+			MountPath: "/fluent-bit/etc/",
+		},
+	}
+	sidecar := logging.CreateFluentBitSideCarContainer("zookeeper", []logging.FluentBitLabel{}, volumeMounts)
+	annotations["zk-fluent-bit.conf"] = utils.Checksum([]byte(fbForwarderConfig["fluent-bit.conf"]))
+	return volume, sidecar
+}
+
 func (r *SFController) DeployZookeeper() bool {
 	dnsNames := r.MkClientDNSNames(zkIdent)
 	privateKey := certv1.CertificatePrivateKey{
@@ -58,15 +96,16 @@ func (r *SFController) DeployZookeeper() bool {
 	cmData := make(map[string]string)
 	cmData["ok.sh"] = zookeeperOk
 	cmData["ready.sh"] = zookeeperReady
-	cmData["run.sh"] = zookeepeRun
+	cmData["run.sh"] = zookeeperRun
+	cmData["logback.xml"] = zkLogbackConfig
 	r.EnsureConfigMap(zkIdent+"-pi", cmData)
 
+	configChecksumable := zookeeperOk + "\n" + zookeeperReady + "\n" + zookeeperRun + "\n" + zkLogbackConfig
+
 	annotations := map[string]string{
-		"ok":     utils.Checksum([]byte(zookeeperOk)),
-		"ready":  utils.Checksum([]byte(zookeeperReady)),
-		"run":    utils.Checksum([]byte(zookeepeRun)),
-		"image":  base.ZookeeperImage,
-		"serial": "1",
+		"configuration": utils.Checksum([]byte(configChecksumable)),
+		"image":         base.ZookeeperImage,
+		"serial":        "2",
 	}
 
 	volumeMounts := []apiv1.VolumeMount{
@@ -92,6 +131,10 @@ func (r *SFController) DeployZookeeper() bool {
 			Name:      zkIdent + "-pi",
 			MountPath: zkPIMountPath,
 		},
+		{
+			Name:      zkIdent + "-logs",
+			MountPath: "/var/log",
+		},
 	}
 
 	srv := base.MkServicePod(zkIdent, r.ns, zkIdent+"-0", []int32{zkSSLPort}, zkIdent)
@@ -101,16 +144,23 @@ func (r *SFController) DeployZookeeper() bool {
 	r.EnsureService(&srvZK)
 
 	storageConfig := r.getStorageConfOrDefault(r.cr.Spec.Zookeeper.Storage)
+	logStorageConfig := base.StorageConfig{
+		Size:             utils.Qty1Gi(),
+		StorageClassName: storageConfig.StorageClassName,
+	}
 	zk := r.mkHeadlessSatefulSet(
 		zkIdent, base.ZookeeperImage, storageConfig, apiv1.ReadWriteOnce)
 	// We overwrite the VolumeClaimTemplates set by MkHeadlessStatefulSet to keep the previous volume name
 	// Previously the default PVC created by MkHeadlessStatefulSet was not used by Zookeeper (not mounted). So to avoid having two Volumes
 	// and to ensure data persistence during the upgrade we keep the previous naming 'zkIdent + "-data"' and we discard the one
 	// created by MkHeadlessStatefulSet.
-	zk.Spec.VolumeClaimTemplates = []apiv1.PersistentVolumeClaim{base.MkPVC(zkIdent+"-data", r.ns, storageConfig, apiv1.ReadWriteOnce)}
+	zk.Spec.VolumeClaimTemplates = []apiv1.PersistentVolumeClaim{
+		base.MkPVC(zkIdent+"-data", r.ns, storageConfig, apiv1.ReadWriteOnce),
+		base.MkPVC(zkIdent+"-logs", r.ns, logStorageConfig, apiv1.ReadWriteOnce),
+	}
 	zk.Spec.Template.Spec.Containers[0].Command = []string{"/bin/bash", "/config-scripts/run.sh"}
 	zk.Spec.Template.Spec.Containers[0].VolumeMounts = volumeMounts
-	zk.Spec.Template.ObjectMeta.Annotations = annotations
+
 	zk.Spec.Template.Spec.Volumes = []apiv1.Volume{
 		base.MkVolumeCM(zkIdent+"-pi", zkIdent+"-pi-config-map"),
 		base.MkVolumeSecret("zookeeper-client-tls"),
@@ -126,8 +176,22 @@ func (r *SFController) DeployZookeeper() bool {
 		base.MkContainerPort(zkServerPort, zkServerPortName),
 	}
 
+	if r.cr.Spec.FluentBitLogForwarding != nil {
+		fbVolume, fbSidecar := createZKLogForwarderSidecar(r, annotations)
+		zk.Spec.Template.Spec.Containers = append(zk.Spec.Template.Spec.Containers, fbSidecar)
+		zk.Spec.Template.Spec.Volumes = append(zk.Spec.Template.Spec.Volumes, fbVolume)
+	}
+	zk.Spec.Template.ObjectMeta.Annotations = annotations
+
 	current := appsv1.StatefulSet{}
 	if r.GetM(zkIdent, &current) {
+		// VolumeClaimTemplates cannot be updated on a statefulset. If we are missing the logs PVC we must
+		// recreate from scratch the statefulset. The new statefulset should mount the old PVC again.
+		if len(current.Spec.VolumeClaimTemplates) < 2 {
+			r.log.V(1).Info("Zookeeper volume claim templates changed, recreating statefulset ...")
+			r.DeleteR(&current)
+			return false
+		}
 		if !utils.MapEquals(&current.Spec.Template.ObjectMeta.Annotations, &annotations) {
 			r.log.V(1).Info("Zookeeper configuration changed, rollout pods ...")
 			current.Spec.Template = zk.DeepCopy().Spec.Template