From 1d638307923a612d67628db80fed0fc5a37e0381 Mon Sep 17 00:00:00 2001 From: Mike Lang Date: Fri, 9 Dec 2016 18:23:55 -0800 Subject: [PATCH] awsecs reporter: Add lots of debug logging and fix bugs - describeServices wasn't describing the partial page left over at the end, which would cause incorrect results - the shim between listServices and describeServices was closing the channel every iteration, which would cause panic for write to closed channel - client was not being saved when created, so it gets recreated each time - we were describeTasks'ing even if we had no tasks to describe --- probe/awsecs/client.go | 98 ++++++++++++++++++++++++++-------------- probe/awsecs/reporter.go | 3 ++ 2 files changed, 68 insertions(+), 33 deletions(-) diff --git a/probe/awsecs/client.go b/probe/awsecs/client.go index 2fd526448..9dfe58a1c 100644 --- a/probe/awsecs/client.go +++ b/probe/awsecs/client.go @@ -111,12 +111,15 @@ func newECSService(service *ecs.Service) ecsService { // Returns a channel from which service ARNs can be read. // Cannot fail as it will attempt to deliver partial results, though that may end up being no results. func (c ecsClient) listServices() <-chan string { + log.Debugf("Listing ECS services") results := make(chan string) go func() { + count := 0 err := c.client.ListServicesPages( &ecs.ListServicesInput{Cluster: &c.cluster}, func(page *ecs.ListServicesOutput, lastPage bool) bool { for _, arn := range page.ServiceArns { + count++ results <- *arn } return true @@ -125,6 +128,7 @@ func (c ecsClient) listServices() <-chan string { if err != nil { log.Warnf("Error listing ECS services, ECS service report may be incomplete: %v", err) } + log.Debugf("Listed %d services", count) close(results) }() return results @@ -136,49 +140,62 @@ func (c ecsClient) describeServices() (chan<- string, <-chan bool) { input := make(chan string) done := make(chan bool) + log.Debugf("Describing ECS services") + go func() { const maxServices = 10 // How many services we can put in one Describe command group := sync.WaitGroup{} lock := sync.Mutex{} // mediates access to the service cache when writing results + describePage := func(arns []string) { + defer group.Done() + + arnPtrs := make([]*string, 0, len(arns)) + for i := range arns { + arnPtrs = append(arnPtrs, &arns[i]) + } + + resp, err := c.client.DescribeServices(&ecs.DescribeServicesInput{ + Cluster: &c.cluster, + Services: arnPtrs, + }) + if err != nil { + log.Warnf("Error describing some ECS services, ECS service report may be incomplete: %v", err) + return + } + + for _, failure := range resp.Failures { + log.Warnf("Failed to describe ECS service %s, ECS service report may be incomplete: %s", *failure.Arn, failure.Reason) + } + + lock.Lock() + for _, service := range resp.Services { + c.serviceCache[*service.ServiceName] = newECSService(service) + } + lock.Unlock() + } + + count := 0 // this is just for logging + calls := 0 // this is just for logging page := make([]string, 0, maxServices) for arn := range input { page = append(page, arn) if len(page) == maxServices { group.Add(1) - - go func(arns []string) { - defer group.Done() - - arnPtrs := make([]*string, 0, len(arns)) - for i := range arns { - arnPtrs = append(arnPtrs, &arns[i]) - } - - resp, err := c.client.DescribeServices(&ecs.DescribeServicesInput{ - Cluster: &c.cluster, - Services: arnPtrs, - }) - if err != nil { - log.Warnf("Error describing some ECS services, ECS service report may be incomplete: %v", err) - return - } - - for _, failure := range resp.Failures { - log.Warnf("Failed to describe ECS service %s, ECS service report may be incomplete: %s", *failure.Arn, failure.Reason) - } - - lock.Lock() - for _, service := range resp.Services { - c.serviceCache[*service.ServiceName] = newECSService(service) - } - lock.Unlock() - }(page) - + go describePage(page) + count += len(page) + calls++ page = make([]string, 0, maxServices) } } + if len(page) > 0 { + group.Add(1) + go describePage(page) + count += len(page) + calls++ + } + log.Debugf("Described %d services in %d calls", count, calls) group.Wait() close(done) }() @@ -188,6 +205,8 @@ func (c ecsClient) describeServices() (chan<- string, <-chan bool) { // get details on given tasks, updating cache with the results func (c ecsClient) getTasks(taskARNs []string) { + log.Debugf("Describing %d ECS tasks", len(taskARNs)) + taskPtrs := make([]*string, len(taskARNs)) for i := range taskARNs { taskPtrs[i] = &taskARNs[i] @@ -218,24 +237,30 @@ func (c ecsClient) evictOldCacheItems() { const evictTime = time.Minute now := time.Now() + count := 0 for arn, task := range c.taskCache { if now.Sub(task.lastUsedAt) > evictTime { delete(c.taskCache, arn) + count++ } } + log.Debugf("Evicted %d old tasks", count) + count = 0 for name, service := range c.serviceCache { if now.Sub(service.lastUsedAt) > evictTime { delete(c.serviceCache, name) + count++ } } + log.Debugf("Evicted %d old services", count) } // Try to match a list of task ARNs to service names using cached info. // Returns (task to service map, unmatched tasks). Ignores tasks whose startedby values // don't appear to point to a service. func (c ecsClient) matchTasksServices(taskARNs []string) (map[string]string, []string) { - const servicePrefix = "aws-svc" // TODO confirm this + const servicePrefix = "ecs-svc" deploymentMap := map[string]string{} for serviceName, service := range c.serviceCache { @@ -243,6 +268,7 @@ func (c ecsClient) matchTasksServices(taskARNs []string) (map[string]string, []s deploymentMap[deployment] = serviceName } } + log.Debugf("Mapped %d deployments from %d services", len(deploymentMap), len(c.serviceCache)) results := map[string]string{} unmatched := []string{} @@ -263,11 +289,13 @@ func (c ecsClient) matchTasksServices(taskARNs []string) (map[string]string, []s } } + log.Debugf("Matched %d from %d tasks, %d unmatched", len(results), len(taskARNs), len(unmatched)) return results, unmatched } // Returns a ecsInfo struct containing data needed for a report. func (c ecsClient) getInfo(taskARNs []string) ecsInfo { + log.Debugf("Getting ECS info on %d tasks", len(taskARNs)) // We do a weird order of operations here to minimize unneeded cache refreshes. // First, we ensure we have all the tasks we need, and fetch the ones we don't. @@ -281,8 +309,10 @@ func (c ecsClient) getInfo(taskARNs []string) ecsInfo { tasksToFetch = append(tasksToFetch, taskARN) } } - // This might not fully succeed, but we only try once and ignore any further missing tasks. - c.getTasks(tasksToFetch) + if len(tasksToFetch) > 0 { + // This might not fully succeed, but we only try once and ignore any further missing tasks. + c.getTasks(tasksToFetch) + } // We're going to do this matching process potentially several times, but that's ok - it's quite cheap. // First, we want to see how far we get with existing data, and identify the set of services @@ -290,6 +320,7 @@ func (c ecsClient) getInfo(taskARNs []string) ecsInfo { taskServiceMap, unmatched := c.matchTasksServices(taskARNs) // In order to ensure service details are fresh, we need to refresh any referenced services + log.Debugf("Refreshing ECS services") toDescribe, done := c.describeServices() servicesRefreshed := map[string]bool{} for _, serviceName := range taskServiceMap { @@ -310,6 +341,7 @@ func (c ecsClient) getInfo(taskARNs []string) ecsInfo { // If we still have tasks unmatched, we'll have to try harder. Get a list of all services and, // if not already refreshed, fetch them. + log.Debugf("After refreshing services, %d tasks unmatched", len(unmatched)) if len(unmatched) > 0 { serviceNamesChan := c.listServices() toDescribe, done := c.describeServices() @@ -319,8 +351,8 @@ func (c ecsClient) getInfo(taskARNs []string) ecsInfo { toDescribe <- serviceName servicesRefreshed[serviceName] = true } - close(toDescribe) } + close(toDescribe) }() <-done diff --git a/probe/awsecs/reporter.go b/probe/awsecs/reporter.go index e8d992616..1f1263355 100644 --- a/probe/awsecs/reporter.go +++ b/probe/awsecs/reporter.go @@ -99,11 +99,13 @@ func (r Reporter) Tag(rpt report.Report) (report.Report, error) { client, ok := r.clients[cluster] if !ok { + log.Debugf("Creating new ECS client") var err error // can't use := on the next line without shadowing outer client var client, err = newClient(cluster) if err != nil { return rpt, err } + r.clients[cluster] = client } taskArns := make([]string, 0, len(taskMap)) @@ -112,6 +114,7 @@ func (r Reporter) Tag(rpt report.Report) (report.Report, error) { } ecsInfo := client.getInfo(taskArns) + log.Debugf("Got info from ECS: %d tasks, %d services", len(ecsInfo.tasks), len(ecsInfo.services)) // Create all the services first for serviceName, service := range ecsInfo.services {