Liveness probe

cmd/mic/main.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/Azure/aad-pod-identity/pkg/mic"
+	"github.com/Azure/aad-pod-identity/pkg/probes"
 	"github.com/Azure/aad-pod-identity/version"
 	"github.com/golang/glog"
 	"k8s.io/client-go/rest"
@@ -19,6 +20,7 @@ var (
 	versionInfo       bool
 	syncRetryDuration time.Duration
 	leaderElectionCfg mic.LeaderElectionConfig
+	httpProbePort     string
 )
 
 func main() {
@@ -39,6 +41,9 @@ func main() {
 	flag.StringVar(&leaderElectionCfg.Name, "leader-election-name", "aad-pod-identity-mic", "leader election name")
 	flag.DurationVar(&leaderElectionCfg.Duration, "leader-election-duration", time.Second*15, "leader election duration")
 
+	//Probe port
+	flag.StringVar(&httpProbePort, "http-probe-port", "8080", "http liveliness probe port")
+
 	flag.Parse()
 	if versionInfo {
 		version.PrintVersionAndExit()
@@ -63,6 +68,12 @@ func main() {
 	if err != nil {
 		glog.Fatalf("Could not get the MIC client: %+v", err)
 	}
+
+	// Health probe will always report success once its started.
+	// MIC instance will report the contents as "Active" only once its elected the leader
+	// and starts the sync loop.
+	probes.InitAndStart(httpProbePort, &micClient.SyncLoopStarted)
+
 	// Starts the leader election loop
 	micClient.Run()
 	glog.Info("AAD Pod identity controller initialized!!")

cmd/nmi/main.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/Azure/aad-pod-identity/pkg/k8s"
 	server "github.com/Azure/aad-pod-identity/pkg/nmi/server"
+	"github.com/Azure/aad-pod-identity/pkg/probes"
 	"github.com/Azure/aad-pod-identity/version"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/pflag"
@@ -28,6 +29,7 @@ var (
 	ipTableUpdateTimeIntervalInSeconds = pflag.Int("ipt-update-interval-sec", defaultIPTableUpdateTimeIntervalInSeconds, "update interval of iptables")
 	forceNamespaced                    = pflag.Bool("forceNamespaced", false, "Forces mic to namespace identities, binding, and assignment")
 	micNamespace                       = pflag.String("MICNamespace", "default", "MIC namespace to short circuit MIC token requests")
+	httpProbePort                      = pflag.String("http-probe-port", "8080", "Http health and liveness probe port")
 )
 
 func main() {
@@ -53,6 +55,10 @@ func main() {
 	s.NodeName = *nodename
 	s.IPTableUpdateTimeIntervalInSeconds = *ipTableUpdateTimeIntervalInSeconds
 
+	// Health probe will always report success once its started. The contents
+	// will report "Active" once the iptables rules are set
+	probes.InitAndStart(*httpProbePort, &s.Initialized)
+
 	if err := s.Run(); err != nil {
 		log.Fatalf("%s", err)
 	}

deploy/infra/master/replicaset/deployment-rbac.yaml

@@ -125,6 +125,12 @@ spec:
         volumeMounts:
         - mountPath: /run/xtables.lock
           name: iptableslock
+        livenessProbe:
+          httpGet:
+            path: /healthz
+            port: 8080
+          initialDelaySeconds: 10
+          periodSeconds: 5
       nodeSelector:
         beta.kubernetes.io/os: linux
 ---
@@ -200,6 +206,12 @@ spec:
         - name: k8s-azure-file
           mountPath: /etc/kubernetes/azure.json
           readOnly: true
+        livenessProbe:
+          httpGet:
+            path: /healthz
+            port: 8080
+          initialDelaySeconds: 10
+          periodSeconds: 5
       volumes:
       - name: k8s-azure-file
         hostPath:

deploy/infra/master/replicaset/deployment.yaml

@@ -81,6 +81,12 @@ spec:
         volumeMounts:
         - mountPath: /run/xtables.lock
           name: iptableslock
+        livenessProbe:
+          httpGet:
+            path: /healthz
+            port: 8080
+          initialDelaySeconds: 10
+          periodSeconds: 5
       nodeSelector:
         beta.kubernetes.io/os: linux
 ---
@@ -116,6 +122,12 @@ spec:
           - name: k8s-azure-file
             mountPath: /etc/kubernetes/azure.json
             readOnly: true
+        livenessProbe:
+          httpGet:
+            path: /healthz
+            port: 8080
+          initialDelaySeconds: 10
+          periodSeconds: 5
       volumes:
       - name: kubeconfig
         hostPath:

pkg/mic/mic.go

@@ -55,6 +55,7 @@ type Client struct {
 	EventChannel      chan aadpodid.EventType
 	NodeClient        NodeGetter
 	IsNamespaced      bool
+	SyncLoopStarted   bool
 	syncRetryInterval time.Duration
 
 	syncing       int32 // protect against conucrrent sync's
@@ -125,7 +126,7 @@ func NewMICClient(cloudconfig string, config *rest.Config, isNamespaced bool, sy
 
 // Run - Initiates the leader election run call to find if its leader and run it
 func (c *Client) Run() {
-	glog.Infof("MIC Leader election initiated")
+	glog.Infof("Initiating MIC Leader election")
 	c.leaderElector.Run()
 }
 
@@ -217,6 +218,7 @@ func (c *Client) Sync(exit <-chan struct{}) {
 	defer ticker.Stop()
 
 	glog.Info("Sync thread started.")
+	c.SyncLoopStarted = true
 	var event aadpodid.EventType
 	for {
 		select {

pkg/nmi/server/server.go

@@ -46,6 +46,7 @@ type Server struct {
 	IPTableUpdateTimeIntervalInSeconds int
 	IsNamespaced                       bool
 	MICNamespace                       string
+	Initialized                        bool
 }
 
 // NMIResponse is the response returned to caller
@@ -80,6 +81,17 @@ func (s *Server) Run() error {
 	return nil
 }
 
+func (s *Server) updateIPTableRulesInternal() {
+	log.Infof("node(%s) hostip(%s) metadataaddress(%s:%s) nmiport(%s)", s.NodeName, s.HostIP, s.MetadataIP, s.MetadataPort, s.NMIPort)
+
+	if err := iptables.AddCustomChain(s.MetadataIP, s.MetadataPort, s.HostIP, s.NMIPort); err != nil {
+		log.Fatalf("%s", err)
+	}
+	if err := iptables.LogCustomChain(); err != nil {
+		log.Fatalf("%s", err)
+	}
+}
+
 // updateIPTableRules ensures the correct iptable rules are set
 // such that metadata requests are received by nmi assigned port
 // NOT originating from HostIP destined to metadata endpoint are
@@ -91,6 +103,11 @@ func (s *Server) updateIPTableRules() {
 	ticker := time.NewTicker(time.Second * time.Duration(s.IPTableUpdateTimeIntervalInSeconds))
 	defer ticker.Stop()
 
+	// Run once before the waiting on ticker for the rules to take effect
+	// immediately.
+	s.updateIPTableRulesInternal()
+	s.Initialized = true
+
 loop:
 	for {
 		select {
@@ -99,13 +116,7 @@ loop:
 			break loop
 
 		case <-ticker.C:
-			log.Infof("node(%s) hostip(%s) metadataaddress(%s:%s) nmiport(%s)", s.NodeName, s.HostIP, s.MetadataIP, s.MetadataPort, s.NMIPort)
-			if err := iptables.AddCustomChain(s.MetadataIP, s.MetadataPort, s.HostIP, s.NMIPort); err != nil {
-				log.Fatalf("%s", err)
-			}
-			if err := iptables.LogCustomChain(); err != nil {
-				log.Fatalf("%s", err)
-			}
+			s.updateIPTableRulesInternal()
 		}
 	}
 }

pkg/probes/probes.go

@@ -0,0 +1,45 @@
+package probes
+
+import (
+	"net/http"
+
+	"github.com/golang/glog"
+)
+
+// InitHealthProbe - sets up a health probe which responds with success (200 - OK) once its initialized.
+// The contents of the healthz endpoint will be the string "Active" if the condition is satisfied.
+// The condition is set to true when the sync cycle has become active in case of MIC and the iptables
+// rules set in case of NMI.
+func InitHealthProbe(condition *bool) {
+	http.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(200)
+		if *condition {
+			w.Write([]byte("Active"))
+		} else {
+			w.Write([]byte("Not Active"))
+		}
+	})
+}
+
+func startAsync(port string) {
+	err := http.ListenAndServe(":"+port, nil)
+	if err != nil {
+		glog.Fatalf("Http listen and serve error: %+v", err)
+	} else {
+		glog.V(1).Infof("Http listen and serve started !")
+	}
+}
+
+//Start - Starts the required http server to start the probe to respond.
+func Start(port string) {
+	go startAsync(port)
+}
+
+// InitAndStart - Initialize the default probes and starts the http listening port.
+func InitAndStart(port string, condition *bool) {
+	InitHealthProbe(condition)
+	glog.V(1).Infof("Initialized health probe")
+	// start the probe.
+	Start(port)
+	glog.Infof("Initialized and started health probe")
+}

test/common/k8s/infra/infra.go

@@ -11,16 +11,22 @@ import (
 )
 
 // CreateInfra will deploy all the infrastructure components (nmi and mic) on a Kubernetes cluster
-func CreateInfra(namespace, registry, nmiVersion, micVersion, templateOutputPath string) error {
-	t, err := template.New("deployment-rbac.yaml").ParseFiles(path.Join("template", "deployment-rbac.yaml"))
+func CreateInfra(namespace, registry, nmiVersion, micVersion, templateOutputPath string, old bool) error {
+	var err error
+	var t *template.Template
+	if !old {
+		t, err = template.New("deployment-rbac.yaml").ParseFiles(path.Join("template", "deployment-rbac.yaml"))
+	} else {
+		t, err = template.New("deployment-rbac-old.yaml").ParseFiles(path.Join("template", "deployment-rbac-old.yaml"))
+	}
 	if err != nil {
 		return errors.Wrap(err, "Failed to parse deployment-rbac.yaml")
 	}
 
 	deployFilePath := path.Join(templateOutputPath, namespace+"-deployment.yaml")
 	deployFile, err := os.Create(deployFilePath)
 	if err != nil {
-		return errors.Wrap(err, "Failed to create a deployment file from deployment-rbac.yaml")
+		return errors.Wrap(err, "Failed to create a deployment file")
 	}
 	defer deployFile.Close()