From a443bf969400448c8e9e68cccbe35928d94590e0 Mon Sep 17 00:00:00 2001
From: Marek Siarkowicz <siarkowicz@google.com>
Date: Wed, 9 May 2018 11:57:41 +0200
Subject: [PATCH] Clean stackdriver sinks when reached limit

---
 .../logging/stackdrvier/utils.go              | 37 +++++++++++++++++++
 1 file changed, 37 insertions(+)

diff --git a/test/e2e/instrumentation/logging/stackdrvier/utils.go b/test/e2e/instrumentation/logging/stackdrvier/utils.go
index 57367b58e3d..27360474a83 100644
--- a/test/e2e/instrumentation/logging/stackdrvier/utils.go
+++ b/test/e2e/instrumentation/logging/stackdrvier/utils.go
@@ -49,6 +49,9 @@ const (
 
 	// The parallelism level of polling logs process.
 	sdLoggingPollParallelism = 10
+
+	// The limit on the number of stackdriver sinks that can be created within one project.
+	stackdriverSinkCountLimit = 90
 )
 
 type logProviderScope int
@@ -86,6 +89,10 @@ func newSdLogProvider(f *framework.Framework, scope logProviderScope) (*sdLogPro
 	if err != nil {
 		return nil, err
 	}
+	err = ensureProjectHasSinkCapacity(sdService.Projects.Sinks, framework.TestContext.CloudConfig.ProjectID)
+	if err != nil {
+		return nil, err
+	}
 
 	pubsubService, err := pubsub.New(hc)
 	if err != nil {
@@ -104,6 +111,36 @@ func newSdLogProvider(f *framework.Framework, scope logProviderScope) (*sdLogPro
 	return provider, nil
 }
 
+func ensureProjectHasSinkCapacity(sinksService *sd.ProjectsSinksService, projectID string) error {
+	listResponse, err := listSinks(sinksService, projectID)
+	if err != nil {
+		return err
+	}
+	if len(listResponse.Sinks) >= stackdriverSinkCountLimit {
+		framework.Logf("Reached Stackdriver sink limit. Deleting all sinks")
+		deleteSinks(sinksService, projectID, listResponse.Sinks)
+	}
+	return nil
+}
+
+func listSinks(sinksService *sd.ProjectsSinksService, projectID string) (*sd.ListSinksResponse, error) {
+	projectDst := fmt.Sprintf("projects/%s", projectID)
+	listResponse, err := sinksService.List(projectDst).PageSize(stackdriverSinkCountLimit).Do()
+	if err != nil {
+		return nil, fmt.Errorf("failed to list Stackdriver Logging sinks: %v", err)
+	}
+	return listResponse, nil
+}
+
+func deleteSinks(sinksService *sd.ProjectsSinksService, projectID string, sinks []*sd.LogSink) {
+	for _, sink := range sinks {
+		sinkNameID := fmt.Sprintf("projects/%s/sinks/%s", projectID, sink.Name)
+		if _, err := sinksService.Delete(sinkNameID).Do(); err != nil {
+			framework.Logf("Failed to delete LogSink: %v", err)
+		}
+	}
+}
+
 func (p *sdLogProvider) Init() error {
 	projectID := framework.TestContext.CloudConfig.ProjectID
 	nsName := p.framework.Namespace.Name