[errors] de-specialize Transient in favor of Tags.

logdog/server/cmd/logdog_collector/main.go

 // Copyright 2016 The LUCI Authors. All rights reserved.
 // Use of this source code is governed under the Apache License, Version 2.0
 // that can be found in the LICENSE file.
 
 package main
 
 import (
         "fmt"
         "time"
 
         "github.com/luci/luci-go/common/clock"
         "github.com/luci/luci-go/common/data/rand/mathrand"
         "github.com/luci/luci-go/common/errors"
         gcps "github.com/luci/luci-go/common/gcloud/pubsub"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/proto/google"
         "github.com/luci/luci-go/common/retry"
+        "github.com/luci/luci-go/common/retry/transient"
         "github.com/luci/luci-go/common/tsmon/distribution"
         "github.com/luci/luci-go/common/tsmon/field"
         "github.com/luci/luci-go/common/tsmon/metric"
         "github.com/luci/luci-go/common/tsmon/types"
         "github.com/luci/luci-go/grpc/grpcutil"
         "github.com/luci/luci-go/hardcoded/chromeinfra"
         "github.com/luci/luci-go/logdog/server/collector"
         "github.com/luci/luci-go/logdog/server/collector/coordinator"
         "github.com/luci/luci-go/logdog/server/service"
 
@@ skipping 115 matching lines @@
                 return &retry.ExponentialBackoff{
                         Limited: retry.Limited{
                                 Delay:   200 * time.Millisecond,
                                 Retries: -1, // Unlimited.
                         },
                         MaxDelay:   10 * time.Second,
                         Multiplier: 2,
                 }
         }
 
-        err = retry.Retry(c, retry.TransientOnly(retryForever), func() error {
+        err = retry.Retry(c, transient.Only(retryForever), func() error {
                 return grpcutil.WrapIfTransient(psSub.Receive(c, func(c context.Context, msg *pubsub.Message) {
                         c = log.SetField(c, "messageID", msg.ID)
                         if a.processMessage(c, &coll, msg) {
                                 // ACK the message, removing it from Pub/Sub.
                                 msg.Ack()
                         } else {
                                 // NACK the message. It will be redelivered and processed.
                                 msg.Nack()
                         }
                 }))
@@ skipping 20 matching lines @@
         }.Infof(c, "Received Pub/Sub Message.")
 
         startTime := clock.Now(c)
         err := coll.Process(c, msg.Data)
         duration := clock.Now(c).Sub(startTime)
 
         // We track processing time in milliseconds.
         tsTaskProcessingTime.Add(c, duration.Seconds()*1000)
 
         switch {
-        case errors.IsTransient(err):
+        case transient.Tag.In(err):
                 // Do not consume
                 log.Fields{
                         log.ErrorKey: err,
                         "duration":   duration,
                 }.Warningf(c, "TRANSIENT error ingesting Pub/Sub message.")
                 tsPubsubCount.Add(c, 1, "transient_failure")
                 return false
 
         case err == nil:
                 log.Fields{
@@ skipping 18 matching lines @@
 func main() {
         mathrand.SeedRandomly()
         a := application{
                 Service: service.Service{
                         Name:               "collector",
                         DefaultAuthOptions: chromeinfra.DefaultAuthOptions(),
                 },
         }
         a.Run(context.Background(), a.runCollector)
 }