Mpuncel/rc audit logs

bin/p2-rctl-server/main.go

@@ -16,12 +16,12 @@ import (
 
 	"github.com/square/p2/pkg/alerting"
 	"github.com/square/p2/pkg/health/checker"
-	"github.com/square/p2/pkg/labels"
 	"github.com/square/p2/pkg/logging"
 	"github.com/square/p2/pkg/rc"
 	"github.com/square/p2/pkg/roll"
 	"github.com/square/p2/pkg/scheduler"
 	"github.com/square/p2/pkg/store/consul"
+	"github.com/square/p2/pkg/store/consul/auditlogstore"
 	"github.com/square/p2/pkg/store/consul/consulutil"
 	"github.com/square/p2/pkg/store/consul/flags"
 	"github.com/square/p2/pkg/store/consul/rcstore"
@@ -54,7 +54,7 @@ func SessionName() string {
 func main() {
 	// Parse custom flags + standard Consul routing options
 	kingpin.Version(version.VERSION)
-	_, opts, _ := flags.ParseWithConsulOptions()
+	_, opts, labeler := flags.ParseWithConsulOptions()
 
 	// Set up the logger
 	logger := logging.NewLogger(logrus.Fields{})
@@ -72,9 +72,6 @@ func main() {
 	httpClient := cleanhttp.DefaultClient()
 	client := consul.NewConsulClient(opts)
 	consulStore := consul.NewConsulStore(client)
-	// flags.ParseWithConsulOptions() because that interface doesn't
-	// support transactions which is now required by the RC store
-	labeler := labels.NewConsulApplicator(client, 0)
 	rcStore := rcstore.NewConsul(client, labeler, RetryCount)
 
 	rollStore := rollstore.NewConsul(client, labeler, nil)
@@ -102,9 +99,12 @@ func main() {
 		}
 	}
 
+	auditLogStore := auditlogstore.NewConsulStore(client.KV())
+
 	// Run the farms!
 	go rc.NewFarm(
 		consulStore,
+		auditLogStore,
 		rcStore,
 		rcStore,
 		rcStore,

bin/p2-rctl/main.go

@@ -22,6 +22,7 @@ import (
 	"github.com/square/p2/pkg/labels"
 	"github.com/square/p2/pkg/logging"
 	"github.com/square/p2/pkg/manifest"
+	pc_fields "github.com/square/p2/pkg/pc/fields"
 	"github.com/square/p2/pkg/rc"
 	"github.com/square/p2/pkg/rc/fields"
 	rc_fields "github.com/square/p2/pkg/rc/fields"
@@ -54,11 +55,13 @@ var (
 	logLevel = kingpin.Flag("log", "Logging level to display.").String()
 	logJSON  = kingpin.Flag("log-json", "Log messages will be JSON formatted").Bool()
 
-	cmdCreate       = kingpin.Command(cmdCreateText, "Create a new replication controller")
-	createManifest  = cmdCreate.Flag("manifest", "manifest file to use for this replication controller").Short('m').Required().String()
-	createNodeSel   = cmdCreate.Flag("node-selector", "node selector that this replication controller should target").Short('n').Required().String()
-	createPodLabels = cmdCreate.Flag("pod-label", "a pod label, in LABEL=VALUE form, to add to this replication controller. Can be specified multiple times.").Short('p').StringMap()
-	createRCLabels  = cmdCreate.Flag("rc-label", "an RC label, in LABEL=VALUE form, to be applied to this replication controller. Can be specified multiple times.").Short('r').StringMap()
+	cmdCreate              = kingpin.Command(cmdCreateText, "Create a new replication controller")
+	createManifest         = cmdCreate.Flag("manifest", "manifest file to use for this replication controller").Short('m').Required().String()
+	createNodeSel          = cmdCreate.Flag("node-selector", "node selector that this replication controller should target").Short('n').Required().String()
+	createPodLabels        = cmdCreate.Flag("pod-label", "a pod label, in LABEL=VALUE form, to add to this replication controller. Can be specified multiple times.").Short('p').StringMap()
+	createRCLabels         = cmdCreate.Flag("rc-label", "an RC label, in LABEL=VALUE form, to be applied to this replication controller. Can be specified multiple times.").Short('r').StringMap()
+	createAvailabilityZone = cmdCreate.Flag("availability-zone", "availability zone that RC should belong to").Short('a').Required().String()
+	createClusterName      = cmdCreate.Flag("cluster-name", "availability zone that RC should belong to").Short('c').Required().String()
 
 	cmdDelete   = kingpin.Command(cmdDeleteText, "Delete a replication controller")
 	deleteID    = cmdDelete.Arg("id", "replication controller uuid to delete").Required().String()
@@ -153,7 +156,14 @@ func main() {
 
 	switch cmd {
 	case cmdCreateText:
-		rctl.Create(*createManifest, *createNodeSel, *createPodLabels, *createRCLabels)
+		rctl.Create(
+			*createManifest,
+			*createNodeSel,
+			pc_fields.AvailabilityZone(*createAvailabilityZone),
+			pc_fields.ClusterName(*createClusterName),
+			*createPodLabels,
+			*createRCLabels,
+		)
 	case cmdDeleteText:
 		rctl.Delete(*deleteID, *deleteForce)
 	case cmdReplicasText:
@@ -197,6 +207,8 @@ type ReplicationControllerStore interface {
 	Create(
 		manifest manifest.Manifest,
 		nodeSelector klabels.Selector,
+		availabilityZone pc_fields.AvailabilityZone,
+		clusterName pc_fields.ClusterName,
 		podLabels klabels.Set,
 		additionalLabels klabels.Set,
 	) (fields.RC, error)
@@ -231,7 +243,14 @@ type rctlParams struct {
 	logger      logging.Logger
 }
 
-func (r rctlParams) Create(manifestPath, nodeSelector string, podLabels map[string]string, rcLabels map[string]string) {
+func (r rctlParams) Create(
+	manifestPath string,
+	nodeSelector string,
+	availabilityZone pc_fields.AvailabilityZone,
+	clusterName pc_fields.ClusterName,
+	podLabels map[string]string,
+	rcLabels map[string]string,
+) {
 	manifest, err := manifest.FromPath(manifestPath)
 	if err != nil {
 		r.logger.WithErrorAndFields(err, logrus.Fields{
@@ -246,7 +265,7 @@ func (r rctlParams) Create(manifestPath, nodeSelector string, podLabels map[stri
 		}).Fatalln("Could not parse node selector")
 	}
 
-	newRC, err := r.rcs.Create(manifest, nodeSel, klabels.Set(podLabels), rcLabels)
+	newRC, err := r.rcs.Create(manifest, nodeSel, availabilityZone, clusterName, klabels.Set(podLabels), rcLabels)
 	if err != nil {
 		r.logger.WithError(err).Fatalln("Could not create replication controller in Consul")
 	}

integration/single-node-slug-deploy/check.go

@@ -428,13 +428,13 @@ func verifyTransferReplicas(errCh chan<- error, tempdir string, logger logging.L
 	builder.SetID("some_pod")
 	man := builder.GetManifest()
 
-	fromRC, err := rcStore.Create(man, klabels.Everything(), nil, nil)
+	fromRC, err := rcStore.Create(man, klabels.Everything(), "some_az", "some_cn", nil, nil)
 	if err != nil {
 		errCh <- util.Errorf("could not create RC for replica transfer test: %s", err)
 		return
 	}
 
-	toRC, err := rcStore.Create(man, klabels.Everything(), nil, nil)
+	toRC, err := rcStore.Create(man, klabels.Everything(), "some_az", "some_cn", nil, nil)
 	if err != nil {
 		errCh <- util.Errorf("could not create second RC for replica transfer test: %s", err)
 		return
@@ -854,7 +854,19 @@ func createHelloReplicationController(dir string) (fields.ID, error) {
 		return "", err
 	}
 
-	cmd := exec.Command("p2-rctl", "--log-json", "create", "--manifest", signedManifestPath, "--node-selector", "test=yes")
+	cmd := exec.Command(
+		"p2-rctl",
+		"--log-json",
+		"create",
+		"--manifest",
+		signedManifestPath,
+		"--node-selector",
+		"test=yes",
+		"--availability-zone",
+		"some_az",
+		"--cluster-name",
+		"some_cn",
+	)
 	out := bytes.Buffer{}
 	cmd.Stdout = &out
 	cmd.Stderr = &out

pkg/audit/rc.go

@@ -0,0 +1,44 @@
+package audit
+
+import (
+	"encoding/json"
+
+	pc_fields "github.com/square/p2/pkg/pc/fields"
+	"github.com/square/p2/pkg/types"
+	"github.com/square/p2/pkg/util"
+)
+
+const (
+	// RcRetargetingEvent represents events in which an RC changes the set of
+	// nodes that it is targeting. This can be used to log the set of nodes that
+	// an RC or pod cluster manages over time
+	RCRetargetingEvent EventType = "REPLICATION_CONTROLLER_RETARGET"
+)
+
+type RCRetargetingDetails struct {
+	PodID            types.PodID                `json:"pod_id"`
+	AvailabilityZone pc_fields.AvailabilityZone `json:"availability_zone"`
+	ClusterName      pc_fields.ClusterName      `json:"cluster_name"`
+	Nodes            []types.NodeName           `json:"nodes"`
+}
+
+func NewRCRetargetingEventDetails(
+	podID types.PodID,
+	az pc_fields.AvailabilityZone,
+	name pc_fields.ClusterName,
+	nodes []types.NodeName,
+) (json.RawMessage, error) {
+	details := RCRetargetingDetails{
+		PodID:            podID,
+		AvailabilityZone: az,
+		ClusterName:      name,
+		Nodes:            nodes,
+	}
+
+	bytes, err := json.Marshal(details)
+	if err != nil {
+		return nil, util.Errorf("could not marshal rc retargeting details as json: %s", err)
+	}
+
+	return json.RawMessage(bytes), nil
+}

pkg/audit/rc_test.go

@@ -0,0 +1,44 @@
+package audit
+
+import (
+	"encoding/json"
+	"reflect"
+	"testing"
+
+	pc_fields "github.com/square/p2/pkg/pc/fields"
+	"github.com/square/p2/pkg/types"
+)
+
+func TestRCRetargetingEventDetails(t *testing.T) {
+	podID := types.PodID("some_pod_id")
+	clusterName := pc_fields.ClusterName("some_cluster_name")
+	az := pc_fields.AvailabilityZone("some_availability_zone")
+	nodes := []types.NodeName{"node1", "node2"}
+
+	detailsJSON, err := NewRCRetargetingEventDetails(podID, az, clusterName, nodes)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var details RCRetargetingDetails
+	err = json.Unmarshal(detailsJSON, &details)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if details.PodID != podID {
+		t.Errorf("expected pod id to be %s but was %s", podID, details.PodID)
+	}
+
+	if details.AvailabilityZone != az {
+		t.Errorf("expected availability zone to be %s but was %s", az, details.AvailabilityZone)
+	}
+
+	if details.ClusterName != clusterName {
+		t.Errorf("expected cluster name to be %s but was %s", clusterName, details.ClusterName)
+	}
+
+	if !reflect.DeepEqual(details.Nodes, nodes) {
+		t.Errorf("expected node list to be %s but was %s", nodes, details.Nodes)
+	}
+}

pkg/labels/consul_applicator.go

@@ -365,6 +365,10 @@ func (c *consulApplicator) RemoveAllLabels(labelType Type, id string) error {
 // the passed transaction rather than synchronously making the requisite consul
 // call
 func (c *consulApplicator) RemoveAllLabelsTxn(ctx context.Context, labelType Type, id string) error {
+	return removeAllLabelsTxn(ctx, labelType, id)
+}
+
+func removeAllLabelsTxn(ctx context.Context, labelType Type, id string) error {
 	path, err := objectPath(labelType, id)
 	if err != nil {
 		return err

pkg/labels/fake_applicator.go

@@ -83,6 +83,10 @@ func (app *fakeApplicator) RemoveLabel(labelType Type, id, name string) error {
 	return nil
 }
 
+func (app *fakeApplicator) RemoveLabelsTxn(ctx context.Context, labelType Type, id string, keysToRemove []string) error {
+	panic("not implemented")
+}
+
 func (app *fakeApplicator) ListLabels(labelType Type) ([]Labeled, error) {
 	res := []Labeled{}
 	for id, set := range app.data[labelType] {

pkg/labels/http_applicator.go

@@ -206,6 +206,10 @@ func (h *httpApplicator) RemoveAllLabels(labelType Type, id string) error {
 	return nil
 }
 
+func (h *httpApplicator) RemoveAllLabelsTxn(ctx context.Context, labelType Type, id string) error {
+	return removeAllLabelsTxn(ctx, labelType, id)
+}
+
 // Finds all labels assigned to all entities under a type
 //
 // GET /api/labels/:type

pkg/labels/label_http_server.go

@@ -1,6 +1,7 @@
 package labels
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
@@ -21,8 +22,11 @@ import (
 type ApplicatorWithoutWatches interface {
 	SetLabel(labelType Type, id, name, value string) error
 	SetLabels(labelType Type, id string, labels map[string]string) error
+	SetLabelsTxn(ctx context.Context, labelType Type, id string, labels map[string]string) error
 	RemoveLabel(labelType Type, id, name string) error
 	RemoveAllLabels(labelType Type, id string) error
+	RemoveLabelsTxn(ctx context.Context, labelType Type, id string, keysToRemove []string) error
+	RemoveAllLabelsTxn(ctx context.Context, labelType Type, id string) error
 	ListLabels(labelType Type) ([]Labeled, error)
 	GetLabels(labelType Type, id string) (Labeled, error)
 	GetMatches(selector klabels.Selector, labelType Type) ([]Labeled, error)

pkg/pc/fields/fields.go

@@ -4,7 +4,6 @@ import (
 	"encoding/json"
 	"reflect"
 
-	"github.com/square/p2/pkg/store/consul/rcstore"
 	"github.com/square/p2/pkg/types"
 
 	"k8s.io/kubernetes/pkg/labels"
@@ -18,9 +17,9 @@ type Annotations map[string]interface{}
 
 // label keys used by pod selector
 const (
-	AvailabilityZoneLabel = "availability_zone"
-	ClusterNameLabel      = "cluster_name"
-	PodIDLabel            = rcstore.PodIDLabel // TODO: put this in a different place now that multiple packages use it
+	AvailabilityZoneLabel = types.AvailabilityZoneLabel
+	ClusterNameLabel      = types.ClusterNameLabel
+	PodIDLabel            = types.PodIDLabel
 )
 
 func (id ID) String() string {