diff --git a/charts/pravega-operator/README.md b/charts/pravega-operator/README.md
index 621663358..39ea1a526 100644
--- a/charts/pravega-operator/README.md
+++ b/charts/pravega-operator/README.md
@@ -66,3 +66,4 @@ The following table lists the configurable parameters of the pravega-operator ch
 | `webhookCert.certName` | Name of the certificate, if generate is set to false | `selfsigned-cert` |
 | `webhookCert.secretName` | Name of the secret created by the certificate, if generate is set to false | `selfsigned-cert-tls` |
 | `watchNamespace` | Namespaces to be watched  | `""` |
+| `logLevel` | Specify the loging level used by operator  | `"debug"` |
diff --git a/charts/pravega-operator/templates/operator.yaml b/charts/pravega-operator/templates/operator.yaml
index afdd92259..e687f1ec0 100644
--- a/charts/pravega-operator/templates/operator.yaml
+++ b/charts/pravega-operator/templates/operator.yaml
@@ -31,6 +31,8 @@ spec:
         args: [-test]
         {{- end }}
         env:
+        - name: LOG_LEVEL
+          value: "{{ .Values.logLevel }}"
         - name: WATCH_NAMESPACE
           value: "{{ .Values.watchNamespace }}"
         - name: POD_NAME
diff --git a/charts/pravega-operator/values.yaml b/charts/pravega-operator/values.yaml
index 0d93fc815..4c6ff75ca 100644
--- a/charts/pravega-operator/values.yaml
+++ b/charts/pravega-operator/values.yaml
@@ -50,6 +50,9 @@ webhookCert:
 ## An empty string means all namespaces.
 watchNamespace: ""
 
+##Specify the loging level used by operator
+logLevel: "debug"
+
 hooks:
   backoffLimit: 10
   image:
diff --git a/cmd/manager/main.go b/cmd/manager/main.go
index 2272fab6e..cc09f1a1c 100644
--- a/cmd/manager/main.go
+++ b/cmd/manager/main.go
@@ -23,21 +23,21 @@ import (
 	"github.com/pravega/pravega-operator/pkg/apis/pravega/v1beta1"
 	"github.com/pravega/pravega-operator/pkg/controller"
 	controllerconfig "github.com/pravega/pravega-operator/pkg/controller/config"
+	"github.com/pravega/pravega-operator/pkg/util"
 	"github.com/pravega/pravega-operator/pkg/version"
-	log "github.com/sirupsen/logrus"
-
+	"github.com/rs/zerolog"
+	log "github.com/rs/zerolog/log"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/oidc"
-
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
-	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
 	"sigs.k8s.io/controller-runtime/pkg/runtime/signals"
 )
 
 var (
 	versionFlag bool
 	webhookFlag bool
+	logLevel    string
 )
 
 func init() {
@@ -47,17 +47,17 @@ func init() {
 }
 
 func printVersion() {
-	log.Printf("pravega-operator Version: %v", version.Version)
-	log.Printf("Git SHA: %s", version.GitSHA)
-	log.Printf("Go Version: %s", runtime.Version())
-	log.Printf("Go OS/Arch: %s/%s", runtime.GOOS, runtime.GOARCH)
-	log.Printf("operator-sdk Version: %v", sdkVersion.Version)
+	log.Info().Msgf("pravega-operator Version: %v", version.Version)
+	log.Info().Msgf("Git SHA: %s", version.GitSHA)
+	log.Info().Msgf("Go Version: %s", runtime.Version())
+	log.Info().Msgf("Go OS/Arch: %s/%s", runtime.GOOS, runtime.GOARCH)
+	log.Info().Msgf("operator-sdk Version: %v", sdkVersion.Version)
 }
 
 func main() {
-	flag.Parse()
-	logf.SetLogger(logf.ZapLogger(false))
+	zerolog.SetGlobalLevel(util.LogLevel())
 
+	flag.Parse()
 	printVersion()
 
 	if versionFlag {
@@ -65,18 +65,18 @@ func main() {
 	}
 
 	if controllerconfig.TestMode {
-		log.Warn("----- Running in test mode. Make sure you are NOT in production -----")
+		log.Warn().Msg("----- Running in test mode. Make sure you are NOT in production -----")
 	}
 
 	namespace, err := k8sutil.GetWatchNamespace()
 	if err != nil {
-		log.Fatal(err, "failed to get watch namespace")
+		log.Error().Err(err).Msg("failed to get watch namespace")
 	}
 
 	// Get a config to talk to the apiserver
 	cfg, err := config.GetConfig()
 	if err != nil {
-		log.Fatal(err)
+		log.Error().Err(err).Msg("")
 	}
 
 	// Become the leader before proceeding
@@ -86,33 +86,42 @@ func main() {
 	mgr, err := manager.New(cfg, manager.Options{Namespace: namespace})
 
 	if err != nil {
-		log.Fatal(err)
+		log.Fatal().
+			Err(err).
+			Msg("")
 	}
 
-	log.Print("Registering Components")
+	log.Info().Msg("Registering Components")
 
 	// Setup Scheme for all resources
 	if err := apis.AddToScheme(mgr.GetScheme()); err != nil {
-		log.Fatal(err)
+		log.Fatal().
+			Err(err).
+			Msg("")
+
 	}
 
 	// Setup all Controllers
 	if err := controller.AddToManager(mgr); err != nil {
-		log.Fatal(err)
+		log.Fatal().
+			Err(err).
+			Msg("")
 	}
 
 	v1beta1.Mgr = mgr
 	if webhookFlag {
 		if err := (&v1beta1.PravegaCluster{}).SetupWebhookWithManager(mgr); err != nil {
-			log.Error(err, "unable to create webhook %s", err.Error())
+			log.Error().Err(err).Msgf("unable to create webhook %s", err.Error())
 			os.Exit(1)
 		}
 	}
 
-	log.Print("Starting the Cmd")
+	log.Info().Msg("Starting the Cmd")
 
 	// Start the Cmd
 	if err := mgr.Start(signals.SetupSignalHandler()); err != nil {
-		log.Fatal(err, "manager exited non-zero")
+		log.Fatal().
+			Err(err).
+			Msg("manager exited non-zero")
 	}
 }
diff --git a/pkg/controller/pravegacluster/pravegacluster_controller.go b/pkg/controller/pravegacluster/pravegacluster_controller.go
index 7822e3c8e..3573f2ffb 100644
--- a/pkg/controller/pravegacluster/pravegacluster_controller.go
+++ b/pkg/controller/pravegacluster/pravegacluster_controller.go
@@ -20,6 +20,8 @@ import (
 	pravegav1beta1 "github.com/pravega/pravega-operator/pkg/apis/pravega/v1beta1"
 	"github.com/pravega/pravega-operator/pkg/controller/pravega"
 	"github.com/pravega/pravega-operator/pkg/util"
+	"github.com/rs/zerolog"
+	log "github.com/rs/zerolog/log"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	policyv1beta1 "k8s.io/api/policy/v1beta1"
@@ -35,8 +37,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
-
-	log "github.com/sirupsen/logrus"
 )
 
 // ReconcileTime is the delay between reconciliations
@@ -56,6 +56,8 @@ func newReconciler(mgr manager.Manager) reconcile.Reconciler {
 // add adds a new Controller to mgr with r as the reconcile.Reconciler
 func add(mgr manager.Manager, r reconcile.Reconciler) error {
 	// Create a new controller
+	zerolog.SetGlobalLevel(util.LogLevel())
+
 	c, err := controller.New("pravegacluster-controller", mgr, controller.Options{Reconciler: r})
 	if err != nil {
 		return err
@@ -86,7 +88,7 @@ type ReconcilePravegaCluster struct {
 // The Controller will requeue the Request to be processed again if the returned error is non-nil or
 // Result.Requeue is true, otherwise upon completion it will remove the work from the queue.
 func (r *ReconcilePravegaCluster) Reconcile(request reconcile.Request) (reconcile.Result, error) {
-	log.Printf("Reconciling PravegaCluster %s/%s\n", request.Namespace, request.Name)
+	log.Info().Msgf("Reconciling PravegaCluster %s", fmt.Sprintf("%s/%s", request.Namespace, request.Name))
 
 	// Fetch the PravegaCluster instance
 	pravegaCluster := &pravegav1beta1.PravegaCluster{}
@@ -96,20 +98,20 @@ func (r *ReconcilePravegaCluster) Reconcile(request reconcile.Request) (reconcil
 			// Request object not found, could have been deleted after reconcile request.
 			// Owned objects are automatically garbage collected. For additional cleanup logic use finalizers.
 			// Return and don't requeue
-			log.Printf("PravegaCluster %s/%s not found. Ignoring since object must be deleted\n", request.Namespace, request.Name)
+			log.Info().Msgf("PravegaCluster %s not found. Ignoring since object must be deleted", fmt.Sprintf("%s/%s", request.Namespace, request.Name))
 			return reconcile.Result{}, nil
 		}
 		// Error reading the object - requeue the request.
-		log.Printf("failed to get PravegaCluster: %v", err)
+		log.Error().Err(err).Msg("failed to get PravegaCluster:")
 		return reconcile.Result{}, err
 	}
 
 	// Set default configuration for unspecified values
 	changed := pravegaCluster.WithDefaults()
 	if changed {
-		log.Printf("Setting default settings for pravega-cluster: %s", request.Name)
+		log.Info().Msgf("Setting default settings for pravega-cluster: %s", request.Name)
 		if err = r.client.Update(context.TODO(), pravegaCluster); err != nil {
-			log.Printf("Error applying defaults on Pravega Cluster %v", err)
+			log.Error().Err(err).Msg("Error applying defaults on Pravega Cluster :")
 			return reconcile.Result{}, err
 		}
 		return reconcile.Result{Requeue: true}, nil
@@ -117,7 +119,7 @@ func (r *ReconcilePravegaCluster) Reconcile(request reconcile.Request) (reconcil
 
 	err = r.run(pravegaCluster)
 	if err != nil {
-		log.Printf("failed to reconcile pravega cluster (%s): %v", pravegaCluster.Name, err)
+		log.Error().Err(err).Msgf("failed to reconcile pravega cluster (%s):", pravegaCluster.Name)
 		return reconcile.Result{}, err
 	}
 	return reconcile.Result{RequeueAfter: ReconcileTime}, nil
@@ -194,7 +196,7 @@ func (r *ReconcilePravegaCluster) reconcileFinalizers(p *pravegav1beta1.PravegaC
 				event := p.NewApplicationEvent("ZKMETA_CLEANUP_ERROR", "ZK Metadata Cleanup Failed", message, "Error")
 				pubErr := r.client.Create(context.TODO(), event)
 				if pubErr != nil {
-					log.Printf("Error publishing zk metadata cleanup failure event to k8s. %v", pubErr)
+					log.Info().Msgf("Error publishing zk metadata cleanup failure event to k8s. %v", pubErr)
 				}
 				return fmt.Errorf(message)
 			}
@@ -435,7 +437,7 @@ func (r *ReconcilePravegaCluster) reconcileSegmentStoreService(p *pravegav1beta1
 							return fmt.Errorf("failed to delete Segmentstore pod (%s) for 10 mins ", pod.Name)
 						}
 						err = r.client.Get(context.TODO(), types.NamespacedName{Name: service.Name, Namespace: p.Namespace}, pod)
-						log.Printf("waiting for %v pod to be deleted", pod.Name)
+						log.Info().Msgf("waiting for %v pod to be deleted", pod.Name)
 					}
 					start = time.Now()
 					err = r.client.Get(context.TODO(), types.NamespacedName{Name: service.Name, Namespace: p.Namespace}, pod)
@@ -444,7 +446,7 @@ func (r *ReconcilePravegaCluster) reconcileSegmentStoreService(p *pravegav1beta1
 							return fmt.Errorf("failed to get Segmentstore pod (%s) as ready for 10 mins ", pod.Name)
 						}
 						err = r.client.Get(context.TODO(), types.NamespacedName{Name: service.Name, Namespace: p.Namespace}, pod)
-						log.Printf("waiting for %v pod to be in ready state", pod.Name)
+						log.Info().Msgf("waiting for %v pod to be in ready state", pod.Name)
 					}
 				}
 			}
@@ -468,7 +470,7 @@ func (r *ReconcilePravegaCluster) cleanUpZookeeperMeta(p *pravegav1beta1.Pravega
 func (r *ReconcilePravegaCluster) deployCluster(p *pravegav1beta1.PravegaCluster) (err error) {
 	err = r.deployController(p)
 	if err != nil {
-		log.Printf("failed to deploy controller: %v", err)
+		log.Error().Err(err).Msg("failed to deploy controller:")
 		return err
 	}
 
@@ -478,7 +480,7 @@ func (r *ReconcilePravegaCluster) deployCluster(p *pravegav1beta1.PravegaCluster
 
 		err = r.deploySegmentStore(p)
 		if err != nil {
-			log.Printf("failed to deploy segment store: %v", err)
+			log.Error().Err(err).Msg("failed to deploy segment store: ")
 			return err
 		}
 
@@ -513,7 +515,7 @@ func (r *ReconcilePravegaCluster) deleteSTS(p *pravegav1beta1.PravegaCluster) er
 	}
 	// delete sts, if found
 	r.client.Delete(context.TODO(), sts)
-	log.Printf("Deleted old SegmentStore STS %s", sts.Name)
+	log.Info().Msgf("Deleted old SegmentStore STS %s", sts.Name)
 	return nil
 }
 
@@ -568,7 +570,7 @@ func (r *ReconcilePravegaCluster) deleteOldSegmentStoreIfExists(p *pravegav1beta
 				return fmt.Errorf("failed to get external service (%s): %v", svcName, err)
 			}
 			r.client.Delete(context.TODO(), extService)
-			log.Printf("Deleted old SegmentStore external service %s", extService)
+			log.Info().Msgf("Deleted old SegmentStore external service %s", extService)
 		}
 	}
 	return nil
@@ -608,7 +610,7 @@ func (r *ReconcilePravegaCluster) deploySegmentStore(p *pravegav1beta1.PravegaCl
 			}
 			owRefs := sts.GetOwnerReferences()
 			if hasOldVersionOwnerReference(owRefs) {
-				log.Printf("Deleting SSS STS as it has old version owner ref.")
+				log.Info().Msgf("Deleting SSS STS as it has old version owner ref.")
 				err = r.client.Delete(context.TODO(), sts)
 				if err != nil {
 					return err
@@ -931,7 +933,7 @@ func (r *ReconcilePravegaCluster) rollbackFailedUpgrade(p *pravegav1beta1.Praveg
 	if r.isRollbackTriggered(p) {
 		// start rollback to previous version
 		previousVersion := p.Status.GetLastVersion()
-		log.Printf("Rolling back to last cluster version  %v", previousVersion)
+		log.Info().Msgf("Rolling back to last cluster version  %v", previousVersion)
 		//Rollback cluster to previous version
 		return r.rollbackClusterVersion(p, previousVersion)
 	}
diff --git a/pkg/controller/pravegacluster/upgrade.go b/pkg/controller/pravegacluster/upgrade.go
index 8df5c0c8f..85965152a 100644
--- a/pkg/controller/pravegacluster/upgrade.go
+++ b/pkg/controller/pravegacluster/upgrade.go
@@ -19,7 +19,8 @@ import (
 	pravegav1beta1 "github.com/pravega/pravega-operator/pkg/apis/pravega/v1beta1"
 	"github.com/pravega/pravega-operator/pkg/controller/pravega"
 	"github.com/pravega/pravega-operator/pkg/util"
-	log "github.com/sirupsen/logrus"
+	"github.com/rs/zerolog"
+	log "github.com/rs/zerolog/log"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
@@ -37,6 +38,8 @@ type componentSyncVersionFun struct {
 
 // upgrade
 func (r *ReconcilePravegaCluster) syncClusterVersion(p *pravegav1beta1.PravegaCluster) (err error) {
+	zerolog.SetGlobalLevel(util.LogLevel())
+
 	defer func() {
 		r.client.Status().Update(context.TODO(), p)
 	}()
@@ -60,25 +63,25 @@ func (r *ReconcilePravegaCluster) syncClusterVersion(p *pravegav1beta1.PravegaCl
 	if upgradeCondition.Status == corev1.ConditionTrue {
 		// Upgrade process already in progress
 		if p.Status.TargetVersion == "" {
-			log.Println("syncing to an unknown version: cancelling upgrade process")
+			log.Info().Msgf("syncing to an unknown version: cancelling upgrade process")
 			return r.clearUpgradeStatus(p)
 		}
 
 		if p.Status.TargetVersion == p.Status.CurrentVersion {
-			log.Printf("syncing to version '%s' completed", p.Status.TargetVersion)
+			log.Info().Msgf("syncing to version '%s' completed", p.Status.TargetVersion)
 			return r.clearUpgradeStatus(p)
 		}
 
 		syncCompleted, err := r.syncComponentsVersion(p)
 		if err != nil {
-			log.Printf("error syncing cluster version, upgrade failed. %v", err)
+			log.Error().Err(err).Msgf("error syncing cluster version, upgrade failed.")
 			p.Status.SetErrorConditionTrue("UpgradeFailed", err.Error())
 			// emit an event for Upgrade Failure
 			message := fmt.Sprintf("Error Upgrading from version %v to %v. %v", p.Status.CurrentVersion, p.Status.TargetVersion, err.Error())
 			event := p.NewEvent("UPGRADE_ERROR", pravegav1beta1.UpgradeErrorReason, message, "Error")
 			pubErr := r.client.Create(context.TODO(), event)
 			if pubErr != nil {
-				log.Printf("Error publishing Upgrade Failure event to k8s. %v", pubErr)
+				log.Info().Msgf("Error publishing Upgrade Failure event to k8s. %v", pubErr)
 			}
 			r.clearUpgradeStatus(p)
 			return err
@@ -88,7 +91,7 @@ func (r *ReconcilePravegaCluster) syncClusterVersion(p *pravegav1beta1.PravegaCl
 			// All component versions have been synced
 			p.Status.AddToVersionHistory(p.Status.TargetVersion)
 			p.Status.CurrentVersion = p.Status.TargetVersion
-			log.Printf("Upgrade completed for all pravega components.")
+			log.Info().Msgf("Upgrade completed for all pravega components.")
 		}
 		return nil
 	}
@@ -103,7 +106,7 @@ func (r *ReconcilePravegaCluster) syncClusterVersion(p *pravegav1beta1.PravegaCl
 		// skip this check when cluster is in RollbackFailed state
 		if readyCondition == nil || readyCondition.Status != corev1.ConditionTrue {
 			r.clearUpgradeStatus(p)
-			log.Print("cannot trigger upgrade if there are unready pods")
+			log.Info().Msgf("cannot trigger upgrade if there are unready pods")
 			return nil
 		}
 	} else {
@@ -112,7 +115,7 @@ func (r *ReconcilePravegaCluster) syncClusterVersion(p *pravegav1beta1.PravegaCl
 	}
 
 	// Need to sync cluster versions
-	log.Printf("syncing cluster version from %s to %s", p.Status.CurrentVersion, p.Spec.Version)
+	log.Info().Msgf("syncing cluster version from %s to %s", p.Status.CurrentVersion, p.Spec.Version)
 	// Setting target version and condition.
 	// The upgrade process will start on the next reconciliation
 	p.Status.TargetVersion = p.Spec.Version
@@ -144,13 +147,13 @@ func (r *ReconcilePravegaCluster) rollbackClusterVersion(p *pravegav1beta1.Prave
 	if rollbackCondition == nil || rollbackCondition.Status != corev1.ConditionTrue {
 		// We're in the first iteration for Rollback
 		// Add Rollback Condition to Cluster Status
-		log.Printf("Updating Target Version to  %v", version)
+		log.Info().Msgf("Updating Target Version to  %v", version)
 		p.Status.TargetVersion = version
 		p.Status.SetRollbackConditionTrue("", "")
 		updateErr := r.client.Status().Update(context.TODO(), p)
 		if updateErr != nil {
 			p.Status.SetRollbackConditionFalse()
-			log.Printf("Error updating cluster: %v", updateErr.Error())
+			log.Info().Msgf("Error updating cluster: %v", updateErr.Error())
 			return fmt.Errorf("Error updating cluster status. %v", updateErr)
 		}
 		return nil
@@ -165,10 +168,10 @@ func (r *ReconcilePravegaCluster) rollbackClusterVersion(p *pravegav1beta1.Prave
 		event := p.NewEvent("ROLLBACK_ERROR", pravegav1beta1.RollbackErrorReason, message, "Error")
 		pubErr := r.client.Create(context.TODO(), event)
 		if pubErr != nil {
-			log.Printf("Error publishing ROLLBACK_ERROR event to k8s. %v", pubErr)
+			log.Info().Msgf("Error publishing ROLLBACK_ERROR event to k8s. %v", pubErr)
 		}
 		r.clearRollbackStatus(p)
-		log.Printf("Error rolling back to cluster version %v. Reason: %v", version, err)
+		log.Info().Msgf("Error rolling back to cluster version %v. Reason: %v", version, err)
 		//r.client.Status().Update(context.TODO(), p)
 		return err
 	}
@@ -179,14 +182,14 @@ func (r *ReconcilePravegaCluster) rollbackClusterVersion(p *pravegav1beta1.Prave
 		// Set Error/UpgradeFailed Condition to 'false', so rollback is not triggered again
 		p.Status.SetErrorConditionFalse()
 		r.clearRollbackStatus(p)
-		log.Printf("Rollback to version %v completed for all pravega components.", version)
+		log.Info().Msgf("Rollback to version %v completed for all pravega components.", version)
 	}
 	//r.client.Status().Update(context.TODO(), p)
 	return nil
 }
 
 func (r *ReconcilePravegaCluster) clearRollbackStatus(p *pravegav1beta1.PravegaCluster) (err error) {
-	log.Printf("clearRollbackStatus")
+	log.Info().Msgf("clearRollbackStatus")
 	p.Status.SetRollbackConditionFalse()
 	p.Status.TargetVersion = ""
 	// need to deep copy the status struct, otherwise it will be overwritten
@@ -217,7 +220,7 @@ func (r *ReconcilePravegaCluster) syncComponentsVersion(p *pravegav1beta1.Praveg
 		startIndex := len(componentSyncFuncs) - 1
 		// update components in reverse order
 		for i := startIndex; i >= 0; i-- {
-			log.Printf("Rollback: syncing component %v", i)
+			log.Info().Msgf("Rollback: syncing component %v", i)
 			component := componentSyncFuncs[i]
 			synced, err := r.syncComponent(component, p)
 			if !synced {
@@ -232,7 +235,7 @@ func (r *ReconcilePravegaCluster) syncComponentsVersion(p *pravegav1beta1.Praveg
 			}
 		}
 	}
-	log.Printf("Version sync completed for all components.")
+	log.Info().Msgf("Version sync completed for all components.")
 	return true, nil
 }
 
@@ -247,7 +250,7 @@ func (r *ReconcilePravegaCluster) syncComponent(component componentSyncVersionFu
 		// Do not continue with the next component until this one is done
 		return false, nil
 	}
-	log.Printf("%s version sync has been completed", component.name)
+	log.Info().Msgf("%s version sync has been completed", component.name)
 	return true, nil
 }
 
@@ -269,7 +272,7 @@ func (r *ReconcilePravegaCluster) syncControllerVersion(p *pravegav1beta1.Praveg
 
 		// Need to update pod template
 		// This will trigger the rolling upgrade process
-		log.Printf("updating deployment (%s) pod template image to '%s'", deploy.Name, targetImage)
+		log.Info().Msgf("updating deployment (%s) pod template image to '%s'", deploy.Name, targetImage)
 
 		configMap := pravega.MakeControllerConfigMap(p)
 		controllerutil.SetControllerReference(p, configMap, r.scheme)
@@ -288,7 +291,7 @@ func (r *ReconcilePravegaCluster) syncControllerVersion(p *pravegav1beta1.Praveg
 	}
 
 	// Pod template already updated
-	log.Printf("deployment (%s) status: %d updated, %d ready, %d target", deploy.Name,
+	log.Info().Msgf("deployment (%s) status: %d updated, %d ready, %d target", deploy.Name,
 		deploy.Status.UpdatedReplicas, deploy.Status.ReadyReplicas, deploy.Status.Replicas)
 
 	// Check whether the upgrade is in progress or has completed
@@ -338,7 +341,7 @@ func (r *ReconcilePravegaCluster) syncSegmentStoreVersion(p *pravegav1beta1.Prav
 		p.Status.UpdateProgress(pravegav1beta1.UpdatingSegmentstoreReason, "0")
 		// Need to update pod template
 		// This will trigger the rolling upgrade process
-		log.Printf("updating statefulset (%s) template image to '%s'", sts.Name, targetImage)
+		log.Info().Msgf("updating statefulset (%s) template image to '%s'", sts.Name, targetImage)
 
 		configMap := pravega.MakeSegmentstoreConfigMap(p)
 		controllerutil.SetControllerReference(p, configMap, r.scheme)
@@ -358,7 +361,7 @@ func (r *ReconcilePravegaCluster) syncSegmentStoreVersion(p *pravegav1beta1.Prav
 	}
 
 	// Pod template already updated
-	log.Printf("statefulset (%s) status: %d updated, %d ready, %d target", sts.Name,
+	log.Info().Msgf("statefulset (%s) status: %d updated, %d ready, %d target", sts.Name,
 		sts.Status.UpdatedReplicas, sts.Status.ReadyReplicas, sts.Status.Replicas)
 	// Check whether the upgrade is in progress or has completed
 	if sts.Status.UpdatedReplicas == sts.Status.Replicas &&
@@ -394,7 +397,7 @@ func (r *ReconcilePravegaCluster) syncSegmentStoreVersion(p *pravegav1beta1.Prav
 			return false, fmt.Errorf("could not obtain outdated pod")
 		}
 
-		log.Infof("upgrading pod: %s", pod.Name)
+		log.Info().Msgf("upgrading pod: %s", pod.Name)
 
 		err = r.client.Delete(context.TODO(), pod)
 		if err != nil {
@@ -423,7 +426,7 @@ func (r *ReconcilePravegaCluster) IsAbove07STSPresent(p *pravegav1beta1.PravegaC
 		if errors.IsNotFound(err) {
 			return false
 		}
-		log.Printf("failed to get StatefulSet: %v", err)
+		log.Error().Err(err).Msg("failed to get StatefulSet:")
 		return false
 	}
 	return true
@@ -488,7 +491,7 @@ func (r *ReconcilePravegaCluster) syncSegmentStoreVersionTo07(p *pravegav1beta1.
 					*newsts.Spec.Replicas = 0
 					err2 := r.client.Create(context.TODO(), newsts)
 					if err2 != nil {
-						log.Printf("failed to create StatefulSet: %v", err2)
+						log.Error().Err(err).Msg("failed to create StatefulSet:")
 						return false, err2
 					}
 					return false, err
@@ -497,11 +500,11 @@ func (r *ReconcilePravegaCluster) syncSegmentStoreVersionTo07(p *pravegav1beta1.
 			*newsts.Spec.Replicas = 0
 			err2 := r.client.Create(context.TODO(), newsts)
 			if err2 != nil {
-				log.Printf("failed to create StatefulSet: %v", err2)
+				log.Error().Err(err).Msg("failed to create StatefulSet:")
 				return false, err2
 			}
 		} else {
-			log.Printf("failed to get StatefulSet: %v", err)
+			log.Error().Err(err).Msg("failed to get StatefulSet: ")
 			return false, err
 		}
 	}
@@ -524,7 +527,7 @@ func (r *ReconcilePravegaCluster) syncSegmentStoreVersionTo07(p *pravegav1beta1.
 				return true, nil
 			}
 		}
-		log.Printf("failed to get StatefulSet: %v", err)
+		log.Error().Err(err).Msg("failed to get StatefulSet: ")
 		return false, err
 	}
 
diff --git a/pkg/util/pravegacluster.go b/pkg/util/pravegacluster.go
index 109db05be..f89a5365c 100644
--- a/pkg/util/pravegacluster.go
+++ b/pkg/util/pravegacluster.go
@@ -12,12 +12,14 @@ package util
 
 import (
 	"fmt"
+	"os"
 	"reflect"
 	"regexp"
 	"strconv"
 	"strings"
 
 	v "github.com/hashicorp/go-version"
+	"github.com/rs/zerolog"
 	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -35,6 +37,19 @@ func init() {
 	versionRegexp = regexp.MustCompile(MajorMinorVersionRegexp)
 }
 
+func LogLevel() zerolog.Level {
+	logLevel, ok := os.LookupEnv("LOG_LEVEL")
+	if !ok {
+		logLevel = "debug"
+	}
+
+	level, err := zerolog.ParseLevel(logLevel)
+	if err != nil {
+		panic(err)
+	}
+	return level
+}
+
 //function to check if the version is below 0.7 or not
 func IsVersionBelow07(ver string) bool {
 	if ver == "" {
diff --git a/pkg/util/pravegacluster_test.go b/pkg/util/pravegacluster_test.go
index 0f04622d3..2b3cf33b8 100644
--- a/pkg/util/pravegacluster_test.go
+++ b/pkg/util/pravegacluster_test.go
@@ -10,6 +10,7 @@
 package util
 
 import (
+	"fmt"
 	"testing"
 
 	. "github.com/onsi/ginkgo"
@@ -47,6 +48,15 @@ var _ = Describe("pravegacluster", func() {
 			Ω(result4).To(Equal(true))
 		})
 	})
+	Context("Testing loglevel Function", func() {
+		var ans string
+		BeforeEach(func() {
+			ans = fmt.Sprintf("%s", LogLevel())
+		})
+		It("should return true for result", func() {
+			Ω(ans).To(Equal("debug"))
+		})
+	})
 	Context("ContainsVersion fn", func() {
 		var result1, result2 bool
 		BeforeEach(func() {