Feature/tnz 75769 emergency tls route unregistration

jobs/rep/templates/drain.erb

@@ -32,6 +32,19 @@ heartbeat() {
   /var/vcap/packages/rep/bin/gocurl ${curlCmd}/ping >/dev/null 2>&1
 }
 
+emergency_unregister_routes() {
+  local re_pid
+  re_pid=$(/var/vcap/jobs/bpm/bin/bpm pid route_emitter 2>/dev/null) || true
+  if [ -n "$re_pid" ]; then
+    echo "$(date +%Y-%m-%dT%H:%M:%S.%sZ): sending SIGUSR2 to route-emitter (pid $re_pid) for emergency route unregistration"
+    kill -USR2 "$re_pid" || true
+    sleep 35
+    echo "$(date +%Y-%m-%dT%H:%M:%S.%sZ): emergency route unregistration window complete"
+  else
+    echo "$(date +%Y-%m-%dT%H:%M:%S.%sZ): route-emitter not running, skipping emergency unregistration"
+  fi
+}
+
 output_for_bosh() {
   exit_code=$?
 
@@ -59,6 +72,8 @@ if evacuate; then
     echo "$(date +%Y-%m-%dT%H:%M:%S.%sZ): waiting"
   done
 
+  emergency_unregister_routes
+
   /var/vcap/jobs/bpm/bin/bpm stop rep
 else
   exit 1

src/code.cloudfoundry.org/route-emitter/cmd/route-emitter/main.go

@@ -8,7 +8,9 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"os/signal"
 	"strconv"
+	"syscall"
 	"time"
 
 	"code.cloudfoundry.org/bbs"
@@ -126,6 +128,15 @@ func main() {
 
 	handler := routehandlers.NewHandler(table, natsEmitter, routingAPIEmitter, localMode, cfg.TCPEnableTLS, metronClient, unregistrationCache)
 
+	usr2Chan := make(chan os.Signal, 1)
+	signal.Notify(usr2Chan, syscall.SIGUSR2)
+	go func() {
+		for range usr2Chan {
+			logger.Info("received-sigusr2-triggering-emergency-unregistration")
+			handler.EmitAllUnregistrations(logger)
+		}
+	}()
+
 	watcher := watcher.NewWatcher(
 		cfg.CellID,
 		bbsClient,

src/code.cloudfoundry.org/route-emitter/routehandlers/handler.go

@@ -331,3 +331,15 @@ func (handler *Handler) emitMessages(logger lager.Logger, messagesToEmit routing
 		}
 	}
 }
+
+func (handler *Handler) EmitAllUnregistrations(logger lager.Logger) {
+	messages := handler.routingTable.GetAllUnregistrationMessages()
+	logger.Info("emergency-route-unregistration", lager.Data{
+		"unregistration-count": len(messages.UnregistrationMessages),
+	})
+	if handler.natsEmitter != nil {
+		if err := handler.natsEmitter.Emit(messages); err != nil {
+			logger.Error("failed-emergency-unregistration", err)
+		}
+	}
+}

src/code.cloudfoundry.org/route-emitter/routehandlers/handler_test.go

@@ -2,6 +2,7 @@ package routehandlers_test
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 
 	mfakes "code.cloudfoundry.org/diego-logging-client/testhelpers"
@@ -1535,4 +1536,55 @@ var _ = Describe("Handler", func() {
 			})
 		})
 	})
+
+	Describe("EmitAllUnregistrations", func() {
+		var dummyUnregMessages routingtable.MessagesToEmit
+
+		BeforeEach(func() {
+			dummyUnregMessages = routingtable.MessagesToEmit{
+				UnregistrationMessages: []routingtable.RegistryMessage{
+					{
+						Host: "1.1.1.1",
+						URIs: []string{"foo.example.com"},
+						Port: 11,
+					},
+				},
+			}
+			fakeTable.GetAllUnregistrationMessagesReturns(dummyUnregMessages)
+		})
+
+		It("calls GetAllUnregistrationMessages on the routing table", func() {
+			routeHandler.EmitAllUnregistrations(logger)
+			Expect(fakeTable.GetAllUnregistrationMessagesCallCount()).To(Equal(1))
+		})
+
+		It("emits unregistration messages via the NATS emitter", func() {
+			routeHandler.EmitAllUnregistrations(logger)
+			Expect(natsEmitter.EmitCallCount()).To(Equal(1))
+			emitted := natsEmitter.EmitArgsForCall(0)
+			Expect(emitted.UnregistrationMessages).To(HaveLen(1))
+			Expect(emitted.UnregistrationMessages[0].Host).To(Equal("1.1.1.1"))
+		})
+
+		Context("when the NATS emitter returns an error", func() {
+			BeforeEach(func() {
+				natsEmitter.EmitReturns(errors.New("nats-down"))
+			})
+
+			It("logs the error and does not panic", func() {
+				Expect(func() { routeHandler.EmitAllUnregistrations(logger) }).NotTo(Panic())
+				Expect(logger.Buffer()).To(gbytes.Say("failed-emergency-unregistration"))
+			})
+		})
+
+		Context("when the NATS emitter is nil", func() {
+			BeforeEach(func() {
+				routeHandler = routehandlers.NewHandler(fakeTable, nil, fakeRoutingAPIEmitter, false, false, fakeMetronClient, fakeUnregistrationCache)
+			})
+
+			It("does not panic", func() {
+				Expect(func() { routeHandler.EmitAllUnregistrations(logger) }).NotTo(Panic())
+			})
+		})
+	})
 })

src/code.cloudfoundry.org/route-emitter/routingtable/routingtable.go

@@ -36,6 +36,7 @@ type RoutingTable interface {
 	Swap(logger lager.Logger, t RoutingTable, domains models.DomainSet) (TCPRouteMappings, MessagesToEmit)
 	GetInternalRoutingEvents() (TCPRouteMappings, MessagesToEmit)
 	GetExternalRoutingEvents() (TCPRouteMappings, MessagesToEmit)
+	GetAllUnregistrationMessages() MessagesToEmit
 
 	// routes
 
@@ -423,6 +424,25 @@ func (t *internalRoutingTable) GetRoutingEvents() (TCPRouteMappings, MessagesToE
 	return mappings, messagesToEmit
 }
 
+func (t *internalRoutingTable) GetAllUnregistrationMessages() MessagesToEmit {
+	t.Lock()
+	defer t.Unlock()
+
+	var messagesToEmit MessagesToEmit
+	for key, route := range t.entries {
+		_, message, _ := t.emitDiffMessages(key, route, RoutableEndpoints{})
+		messagesToEmit = messagesToEmit.Merge(message)
+	}
+	return messagesToEmit
+}
+
+func (t *routingTable) GetAllUnregistrationMessages() MessagesToEmit {
+	http := t.httpRoutesRoutingTable.GetAllUnregistrationMessages()
+	tcp := t.tcpRoutesRoutingTable.GetAllUnregistrationMessages()
+	internal := t.internalRoutesRoutingTable.GetAllUnregistrationMessages()
+	return http.Merge(tcp).Merge(internal)
+}
+
 type routeMapping interface {
 	MessageFor(endpoint Endpoint, directInstanceAddress, emitEndpointUpdatedAt bool) (*RegistryMessage, *tcpmodels.TcpRouteMapping, *RegistryMessage)
 	Hash() interface{}

src/code.cloudfoundry.org/route-emitter/routingtable/routingtable_test.go

@@ -1181,4 +1181,58 @@ var _ = Describe("RoutingTable", func() {
 			})
 		})
 	})
+
+	Describe("GetAllUnregistrationMessages", func() {
+		Context("when the table is empty", func() {
+			It("returns empty MessagesToEmit", func() {
+				result := table.GetAllUnregistrationMessages()
+				Expect(result.UnregistrationMessages).To(BeEmpty())
+				Expect(result.RegistrationMessages).To(BeEmpty())
+			})
+		})
+
+		Context("when the table has an HTTP route and endpoint", func() {
+			BeforeEach(func() {
+				desiredLRP := createDesiredLRP(key.ProcessGUID, int32(1), key.ContainerPort, logGuid, "", *currentTag, runInfo, hostname1)
+				table.SetRoutes(logger, nil, desiredLRP)
+				actualLRP := createActualLRP(key, endpoint1, domain)
+				table.AddEndpoint(logger, actualLRP)
+			})
+
+			It("returns unregistration messages for all active routes", func() {
+				result := table.GetAllUnregistrationMessages()
+				Expect(result.UnregistrationMessages).NotTo(BeEmpty())
+				var hosts []string
+				for _, msg := range result.UnregistrationMessages {
+					hosts = append(hosts, msg.URIs...)
+				}
+				Expect(hosts).To(ContainElement(hostname1))
+			})
+
+			It("returns no registration messages", func() {
+				result := table.GetAllUnregistrationMessages()
+				Expect(result.RegistrationMessages).To(BeEmpty())
+			})
+		})
+
+		Context("when the table has multiple HTTP routes", func() {
+			hostname2 := "bar.example.com"
+
+			BeforeEach(func() {
+				desiredLRP := createDesiredLRP(key.ProcessGUID, int32(1), key.ContainerPort, logGuid, "", *currentTag, runInfo, hostname1, hostname2)
+				table.SetRoutes(logger, nil, desiredLRP)
+				actualLRP := createActualLRP(key, endpoint1, domain)
+				table.AddEndpoint(logger, actualLRP)
+			})
+
+			It("returns unregistration messages for all routes", func() {
+				result := table.GetAllUnregistrationMessages()
+				var uris []string
+				for _, msg := range result.UnregistrationMessages {
+					uris = append(uris, msg.URIs...)
+				}
+				Expect(uris).To(ConsistOf(hostname1, hostname2))
+			})
+		})
+	})
 })