diff --git a/cmd/allocator/main.go b/cmd/allocator/main.go
index 1f31bc6120..18d466ad0e 100644
--- a/cmd/allocator/main.go
+++ b/cmd/allocator/main.go
@@ -88,64 +88,66 @@ func main() {
 		return err
 	})
 
-	h := newServiceHandler(kubeClient, agonesClient, health)
-
-	// creates a new file watcher for client certificate folder
-	watcher, err := fsnotify.NewWatcher()
-	if err != nil {
-		logger.WithError(err).Fatal("could not create watcher for client certs")
-	}
-	defer watcher.Close() // nolint: errcheck
-	if err := watcher.Add(certDir); err != nil {
-		logger.WithError(err).Fatalf("cannot watch folder %s for secret changes", certDir)
-	}
-
-	watcherTLS, err := fsnotify.NewWatcher()
-	if err != nil {
-		logger.WithError(err).Fatal("could not create watcher for tls certs")
-	}
-	defer watcherTLS.Close() // nolint: errcheck
-	if err := watcherTLS.Add(tlsDir); err != nil {
-		logger.WithError(err).Fatalf("cannot watch folder %s for secret changes", tlsDir)
-	}
+	h := newServiceHandler(kubeClient, agonesClient, health, conf.MTLSDisabled)
 
 	listener, err := net.Listen("tcp", fmt.Sprintf(":%s", sslPort))
 	if err != nil {
 		logger.WithError(err).Fatalf("failed to listen on TCP port %s", sslPort)
 	}
 
-	// Watching for the events in certificate directory for updating certificates, when there is a change
-	go func() {
-		for {
-			select {
-			// watch for events
-			case event := <-watcherTLS.Events:
-				tlsCert, err := readTLSCert()
-				if err != nil {
-					logger.WithError(err).Error("could not load TLS cert; keeping old one")
-				} else {
-					h.tlsMutex.Lock()
-					h.tlsCert = tlsCert
-					h.tlsMutex.Unlock()
-				}
-				logger.Infof("Tls directory change event %v", event)
-			case event := <-watcher.Events:
-				h.certMutex.Lock()
-				caCertPool, err := getCACertPool(certDir)
-				if err != nil {
-					logger.WithError(err).Error("could not load CA certs; keeping old ones")
-				} else {
-					h.caCertPool = caCertPool
-				}
-				logger.Infof("Certificate directory change event %v", event)
-				h.certMutex.Unlock()
+	if !h.mTLSDisabled {
+		// creates a new file watcher for client certificate folder
+		watcher, err := fsnotify.NewWatcher()
+		if err != nil {
+			logger.WithError(err).Fatal("could not create watcher for client certs")
+		}
+		defer watcher.Close() // nolint: errcheck
+		if err := watcher.Add(certDir); err != nil {
+			logger.WithError(err).Fatalf("cannot watch folder %s for secret changes", certDir)
+		}
 
-				// watch for errors
-			case err := <-watcher.Errors:
-				logger.WithError(err).Error("error watching for certificate directory")
-			}
+		watcherTLS, err := fsnotify.NewWatcher()
+		if err != nil {
+			logger.WithError(err).Fatal("could not create watcher for tls certs")
 		}
-	}()
+		defer watcherTLS.Close() // nolint: errcheck
+		if err := watcherTLS.Add(tlsDir); err != nil {
+			logger.WithError(err).Fatalf("cannot watch folder %s for secret changes", tlsDir)
+		}
+
+		// Watching for the events in certificate directory for updating certificates, when there is a change
+		go func() {
+			for {
+				select {
+				// watch for events
+				case event := <-watcherTLS.Events:
+					tlsCert, err := readTLSCert()
+					if err != nil {
+						logger.WithError(err).Error("could not load TLS cert; keeping old one")
+					} else {
+						h.tlsMutex.Lock()
+						h.tlsCert = tlsCert
+						h.tlsMutex.Unlock()
+					}
+					logger.Infof("Tls directory change event %v", event)
+				case event := <-watcher.Events:
+					h.certMutex.Lock()
+					caCertPool, err := getCACertPool(certDir)
+					if err != nil {
+						logger.WithError(err).Error("could not load CA certs; keeping old ones")
+					} else {
+						h.caCertPool = caCertPool
+					}
+					logger.Infof("Certificate directory change event %v", event)
+					h.certMutex.Unlock()
+
+					// watch for errors
+				case err := <-watcher.Errors:
+					logger.WithError(err).Error("error watching for certificate directory")
+				}
+			}
+		}()
+	}
 
 	opts := h.getServerOptions()
 
@@ -165,7 +167,7 @@ func main() {
 	logger.WithError(err).Fatal("allocation service crashed")
 }
 
-func newServiceHandler(kubeClient kubernetes.Interface, agonesClient versioned.Interface, health healthcheck.Handler) *serviceHandler {
+func newServiceHandler(kubeClient kubernetes.Interface, agonesClient versioned.Interface, health healthcheck.Handler, mTLSDisabled bool) *serviceHandler {
 	defaultResync := 30 * time.Second
 	agonesInformerFactory := externalversions.NewSharedInformerFactory(agonesClient, defaultResync)
 	kubeInformerFactory := informers.NewSharedInformerFactory(kubeClient, defaultResync)
@@ -182,6 +184,7 @@ func newServiceHandler(kubeClient kubernetes.Interface, agonesClient versioned.I
 		allocationCallback: func(gsa *allocationv1.GameServerAllocation) (k8sruntime.Object, error) {
 			return allocator.Allocate(gsa, stop)
 		},
+		mTLSDisabled: mTLSDisabled,
 	}
 
 	kubeInformerFactory.Start(stop)
@@ -190,21 +193,23 @@ func newServiceHandler(kubeClient kubernetes.Interface, agonesClient versioned.I
 		logger.WithError(err).Fatal("starting allocator failed.")
 	}
 
-	caCertPool, err := getCACertPool(certDir)
-	if err != nil {
-		logger.WithError(err).Fatal("could not load CA certs.")
-	}
-	h.certMutex.Lock()
-	h.caCertPool = caCertPool
-	h.certMutex.Unlock()
+	if !h.mTLSDisabled {
+		caCertPool, err := getCACertPool(certDir)
+		if err != nil {
+			logger.WithError(err).Fatal("could not load CA certs.")
+		}
+		h.certMutex.Lock()
+		h.caCertPool = caCertPool
+		h.certMutex.Unlock()
 
-	tlsCert, err := readTLSCert()
-	if err != nil {
-		logger.WithError(err).Fatal("could not load TLS certs.")
+		tlsCert, err := readTLSCert()
+		if err != nil {
+			logger.WithError(err).Fatal("could not load TLS certs.")
+		}
+		h.tlsMutex.Lock()
+		h.tlsCert = tlsCert
+		h.tlsMutex.Unlock()
 	}
-	h.tlsMutex.Lock()
-	h.tlsCert = tlsCert
-	h.tlsMutex.Unlock()
 
 	return &h
 }
@@ -220,6 +225,9 @@ func readTLSCert() (*tls.Certificate, error) {
 // getServerOptions returns a list of GRPC server options.
 // Current options are TLS certs and opencensus stats handler.
 func (h *serviceHandler) getServerOptions() []grpc.ServerOption {
+	if h.mTLSDisabled {
+		return []grpc.ServerOption{grpc.StatsHandler(&ocgrpc.ServerHandler{})}
+	}
 
 	cfg := &tls.Config{
 		GetCertificate:        h.getTLSCert,
@@ -323,6 +331,8 @@ type serviceHandler struct {
 
 	tlsMutex sync.RWMutex
 	tlsCert  *tls.Certificate
+
+	mTLSDisabled bool
 }
 
 // Allocate implements the Allocate gRPC method definition
diff --git a/cmd/allocator/metrics.go b/cmd/allocator/metrics.go
index 09ad8b2dbc..7580efa307 100644
--- a/cmd/allocator/metrics.go
+++ b/cmd/allocator/metrics.go
@@ -32,6 +32,7 @@ const (
 	enablePrometheusMetricsFlag  = "prometheus-exporter"
 	projectIDFlag                = "gcp-project-id"
 	stackdriverLabels            = "stackdriver-labels"
+	mTLSDisabledFlag             = "disable-mtls"
 )
 
 func init() {
@@ -39,6 +40,7 @@ func init() {
 }
 
 type config struct {
+	MTLSDisabled      bool
 	PrometheusMetrics bool
 	Stackdriver       bool
 	GCPProjectID      string
@@ -51,11 +53,13 @@ func parseEnvFlags() config {
 	viper.SetDefault(enableStackdriverMetricsFlag, false)
 	viper.SetDefault(projectIDFlag, "")
 	viper.SetDefault(stackdriverLabels, "")
+	viper.SetDefault(mTLSDisabledFlag, false)
 
 	pflag.Bool(enablePrometheusMetricsFlag, viper.GetBool(enablePrometheusMetricsFlag), "Flag to activate metrics of Agones. Can also use PROMETHEUS_EXPORTER env variable.")
 	pflag.Bool(enableStackdriverMetricsFlag, viper.GetBool(enableStackdriverMetricsFlag), "Flag to activate stackdriver monitoring metrics for Agones. Can also use STACKDRIVER_EXPORTER env variable.")
 	pflag.String(projectIDFlag, viper.GetString(projectIDFlag), "GCP ProjectID used for Stackdriver, if not specified ProjectID from Application Default Credentials would be used. Can also use GCP_PROJECT_ID env variable.")
 	pflag.String(stackdriverLabels, viper.GetString(stackdriverLabels), "A set of default labels to add to all stackdriver metrics generated. By default metadata are automatically added using Kubernetes API and GCP metadata enpoint.")
+	pflag.Bool(mTLSDisabledFlag, viper.GetBool(mTLSDisabledFlag), "Flag to enable/disable mTLS in the allocator.")
 	runtime.FeaturesBindFlags()
 	pflag.Parse()
 
@@ -64,6 +68,7 @@ func parseEnvFlags() config {
 	runtime.Must(viper.BindEnv(enableStackdriverMetricsFlag))
 	runtime.Must(viper.BindEnv(projectIDFlag))
 	runtime.Must(viper.BindEnv(stackdriverLabels))
+	runtime.Must(viper.BindEnv(mTLSDisabledFlag))
 	runtime.Must(viper.BindPFlags(pflag.CommandLine))
 	runtime.Must(runtime.FeaturesBindEnv())
 
@@ -74,6 +79,7 @@ func parseEnvFlags() config {
 		Stackdriver:       viper.GetBool(enableStackdriverMetricsFlag),
 		GCPProjectID:      viper.GetString(projectIDFlag),
 		StackdriverLabels: viper.GetString(stackdriverLabels),
+		MTLSDisabled:      viper.GetBool(mTLSDisabledFlag),
 	}
 }
 
diff --git a/install/helm/agones/templates/service/allocation.yaml b/install/helm/agones/templates/service/allocation.yaml
index b2db0cfd5a..c25380ce95 100644
--- a/install/helm/agones/templates/service/allocation.yaml
+++ b/install/helm/agones/templates/service/allocation.yaml
@@ -118,6 +118,8 @@ spec:
           value: {{ .Values.agones.metrics.stackdriverProjectID | quote }}
         - name: STACKDRIVER_LABELS
           value: {{ .Values.agones.metrics.stackdriverLabels | quote }}
+        - name: DISABLE_MTLS
+          value: {{ .Values.agones.allocator.disableMTLS | quote }}
         - name: POD_NAME
           valueFrom:
             fieldRef:
diff --git a/install/helm/agones/values.yaml b/install/helm/agones/values.yaml
index 45d10f7ea8..d3beb5b25d 100644
--- a/install/helm/agones/values.yaml
+++ b/install/helm/agones/values.yaml
@@ -126,6 +126,7 @@ agones:
       serviceType: LoadBalancer
       annotations: {}
     generateTLS: true
+    disableMTLS: false
   image:
     registry: gcr.io/agones-images
     tag: 1.7.0
diff --git a/install/yaml/install.yaml b/install/yaml/install.yaml
index dc5994728a..bb9f4e3ece 100644
--- a/install/yaml/install.yaml
+++ b/install/yaml/install.yaml
@@ -1574,6 +1574,8 @@ spec:
           value: ""
         - name: STACKDRIVER_LABELS
           value: ""
+        - name: DISABLE_MTLS
+          value: "false"
         - name: POD_NAME
           valueFrom:
             fieldRef:
diff --git a/pkg/gameserverallocations/allocator.go b/pkg/gameserverallocations/allocator.go
index 44bcbebdd8..c977f366e6 100644
--- a/pkg/gameserverallocations/allocator.go
+++ b/pkg/gameserverallocations/allocator.go
@@ -364,6 +364,7 @@ func (c *Allocator) allocateFromRemoteCluster(gsa *allocationv1.GameServerAlloca
 
 // createRemoteClusterDialOption creates a grpc client dial option with proper certs to make a remote call.
 func (c *Allocator) createRemoteClusterDialOption(namespace string, connectionInfo *multiclusterv1.ClusterConnectionInfo) (grpc.DialOption, error) {
+	// TODO: disableMTLS works for a single cluster; still need to address how the flag interacts with multi-cluster authentication.
 	clientCert, clientKey, caCert, err := c.getClientCertificates(namespace, connectionInfo.SecretName)
 	if err != nil {
 		return nil, err