Merge pull request #73 from kayac/fix/fcm-error-handling

FCM v1 error handling

config/config.go

@@ -78,6 +78,7 @@ type SectionFCMv1 struct {
 	Enabled                      bool
 	ProjectID                    string
 	TokenSource                  oauth2.TokenSource
+	Endpoint                     string
 }
 
 // DefaultLoadConfig loads default /etc/gunfish.toml

const.go

@@ -78,3 +78,5 @@ var (
 	OutputHookStdout bool
 	OutputHookStderr bool
 )
+
+var RetryBackoff = true

fcmv1/client.go

@@ -46,14 +46,14 @@ func (c *Client) Send(p Payload) ([]Result, error) {
 
 	if body.Error == nil && body.Name != "" {
 		return []Result{
-			Result{
+			{
 				StatusCode: res.StatusCode,
 				Token:      p.Message.Token,
 			},
 		}, nil
 	} else if body.Error != nil {
 		return []Result{
-			Result{
+			{
 				StatusCode: res.StatusCode,
 				Token:      p.Message.Token,
 				Error:      body.Error,
@@ -70,22 +70,28 @@ func (c *Client) NewRequest(p Payload) (*http.Request, error) {
 	if err != nil {
 		return nil, err
 	}
-	token, err := c.tokenSource.Token()
-	if err != nil {
-		return nil, err
+	var bearer string
+	if ts := c.tokenSource; ts != nil {
+		token, err := c.tokenSource.Token()
+		if err != nil {
+			return nil, err
+		}
+		bearer = token.AccessToken
+	} else {
+		bearer = p.Message.Token
 	}
 	req, err := http.NewRequest("POST", c.endpoint.String(), bytes.NewReader(data))
 	if err != nil {
 		return nil, err
 	}
-	req.Header.Set("Authorization", "Bearer "+token.AccessToken)
+	req.Header.Set("Authorization", "Bearer "+bearer)
 	req.Header.Set("Content-Type", "application/json")
 
 	return req, nil
 }
 
 // NewClient establishes a http connection with fcm v1
-func NewClient(tokenSource oauth2.TokenSource, projectID string, endpoint *url.URL, timeout time.Duration) (*Client, error) {
+func NewClient(tokenSource oauth2.TokenSource, projectID string, endpoint string, timeout time.Duration) (*Client, error) {
 	client := &http.Client{
 		Timeout: timeout,
 	}
@@ -94,16 +100,15 @@ func NewClient(tokenSource oauth2.TokenSource, projectID string, endpoint *url.U
 		tokenSource: tokenSource,
 	}
 
-	if endpoint != nil {
-		c.endpoint = endpoint
-	} else {
-		ep, err := url.Parse(DefaultFCMEndpoint)
-		if err != nil {
-			return nil, err
-		}
-		ep.Path = path.Join(ep.Path, projectID, "messages:send")
-		c.endpoint = ep
+	if endpoint == "" {
+		endpoint = DefaultFCMEndpoint
+	}
+	ep, err := url.Parse(endpoint)
+	if err != nil {
+		return nil, err
 	}
+	ep.Path = path.Join(ep.Path, projectID, "messages:send")
+	c.endpoint = ep
 
 	return c, nil
 }

fcmv1/error.go

@@ -9,6 +9,9 @@ const (
 	InvalidArgument = "INVALID_ARGUMENT"
 	Unregistered    = "UNREGISTERED"
 	NotFound        = "NOT_FOUND"
+	Internal        = "INTERNAL"
+	Unavailable     = "UNAVAILABLE"
+	QuotaExceeded   = "QUOTA_EXCEEDED"
 )
 
 type Error struct {

mock/fcmv1_server.go

@@ -0,0 +1,64 @@
+package mock
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"math/rand"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/kayac/Gunfish/fcmv1"
+)
+
+func FCMv1MockServer(projectID string, verbose bool) *http.ServeMux {
+	mux := http.NewServeMux()
+	p := fmt.Sprintf("/v1/projects/%s/messages:send", projectID)
+	log.Println("fcmv1 mock server path:", p)
+	mux.HandleFunc(p, func(w http.ResponseWriter, r *http.Request) {
+		start := time.Now()
+		defer func() {
+			if verbose {
+				log.Printf("reqtime:%f proto:%s method:%s path:%s host:%s", reqtime(start), r.Proto, r.Method, r.URL.Path, r.RemoteAddr)
+			}
+		}()
+
+		// sets the response time from FCM server
+		time.Sleep(time.Millisecond*200 + time.Millisecond*(time.Duration(rand.Int63n(200)-100)))
+		token := r.Header.Get("Authorization")
+		token = strings.TrimPrefix(token, "Bearer ")
+
+		w.Header().Set("Content-Type", ApplicationJSON)
+		switch token {
+		case fcmv1.InvalidArgument:
+			createFCMv1ErrorResponse(w, http.StatusBadRequest, fcmv1.InvalidArgument)
+		case fcmv1.Unregistered:
+			createFCMv1ErrorResponse(w, http.StatusNotFound, fcmv1.Unregistered)
+		case fcmv1.Unavailable:
+			createFCMv1ErrorResponse(w, http.StatusServiceUnavailable, fcmv1.Unavailable)
+		case fcmv1.Internal:
+			createFCMv1ErrorResponse(w, http.StatusInternalServerError, fcmv1.Internal)
+		case fcmv1.QuotaExceeded:
+			createFCMv1ErrorResponse(w, http.StatusTooManyRequests, fcmv1.QuotaExceeded)
+		default:
+			enc := json.NewEncoder(w)
+			enc.Encode(fcmv1.ResponseBody{
+				Name: "ok",
+			})
+		}
+	})
+
+	return mux
+}
+
+func createFCMv1ErrorResponse(w http.ResponseWriter, code int, status string) error {
+	w.WriteHeader(code)
+	enc := json.NewEncoder(w)
+	return enc.Encode(fcmv1.ResponseBody{
+		Error: &fcmv1.FCMError{
+			Status:  status,
+			Message: "mock error:" + status,
+		},
+	})
+}

supervisor.go

@@ -5,7 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"net/http"
+	"math"
 	"os"
 	"os/exec"
 	"sync"
@@ -33,15 +33,13 @@ type Supervisor struct {
 
 // Worker sends notification to apns.
 type Worker struct {
-	ac             *apns.Client
-	fcv1           *fcmv1.Client
-	queue          chan Request
-	respq          chan SenderResponse
-	wgrp           *sync.WaitGroup
-	sn             int
-	id             int
-	errorHandler   func(Request, *http.Response, error)
-	successHandler func(Request, *http.Response)
+	ac    *apns.Client
+	fcv1  *fcmv1.Client
+	queue chan Request
+	respq chan SenderResponse
+	wgrp  *sync.WaitGroup
+	sn    int
+	id    int
 }
 
 // SenderResponse is responses to worker from sender.
@@ -113,17 +111,22 @@ func StartSupervisor(conf *config.Config) (Supervisor, error) {
 				for cnt := 0; cnt < RetryOnceCount; cnt++ {
 					select {
 					case req := <-s.retryq:
-						reqs := &[]Request{req}
-						select {
-						case s.queue <- reqs:
-							LogWithFields(logrus.Fields{"type": "retry", "resend_cnt": req.Tries}).
-								Debugf("Enqueue to retry to send notification.")
-						default:
-							LogWithFields(logrus.Fields{"type": "retry"}).
-								Infof("Could not retry to enqueue because the supervisor queue is full.")
+						var delay time.Duration
+						if RetryBackoff {
+							delay = time.Duration(math.Pow(float64(req.Tries), 2)) * 100 * time.Millisecond
 						}
+						time.AfterFunc(delay, func() {
+							reqs := &[]Request{req}
+							select {
+							case s.queue <- reqs:
+								LogWithFields(logrus.Fields{"delay": delay, "type": "retry", "resend_cnt": req.Tries}).
+									Debugf("Enqueue to retry to send notification.")
+							default:
+								LogWithFields(logrus.Fields{"delay": delay, "type": "retry"}).
+									Infof("Could not retry to enqueue because the supervisor queue is full.")
+							}
+						})
 					default:
-						break
 					}
 				}
 			case <-s.exit:
@@ -172,7 +175,7 @@ func StartSupervisor(conf *config.Config) (Supervisor, error) {
 			return Supervisor{}, errors.New("FCM legacy is not supported")
 		}
 		if conf.FCMv1.Enabled {
-			fcv1, err = fcmv1.NewClient(conf.FCMv1.TokenSource, conf.FCMv1.ProjectID, nil, fcmv1.ClientTimeout)
+			fcv1, err = fcmv1.NewClient(conf.FCMv1.TokenSource, conf.FCMv1.ProjectID, conf.FCMv1.Endpoint, fcmv1.ClientTimeout)
 			if err != nil {
 				LogWithFields(logrus.Fields{
 					"type": "supervisor",
@@ -192,7 +195,7 @@ func StartSupervisor(conf *config.Config) (Supervisor, error) {
 
 		s.workers = append(s.workers, &worker)
 		s.wgrp.Add(1)
-		go s.spawnWorker(worker, conf)
+		go s.spawnWorker(worker)
 		LogWithFields(logrus.Fields{
 			"type":      "worker",
 			"worker_id": i,
@@ -245,7 +248,7 @@ func (s *Supervisor) Shutdown() {
 	}).Infoln("Stoped supervisor.")
 }
 
-func (s *Supervisor) spawnWorker(w Worker, conf *config.Config) {
+func (s *Supervisor) spawnWorker(w Worker) {
 	atomic.AddInt64(&(srvStats.Workers), 1)
 	defer func() {
 		atomic.AddInt64(&(srvStats.Workers), -1)
@@ -367,23 +370,7 @@ func handleFCMResponse(resp SenderResponse, retryq chan<- Request, cmdq chan Com
 	if resp.Err != nil {
 		req := resp.Req
 		LogWithFields(logf).Warnf("response is nil. reason: %s", resp.Err.Error())
-		if req.Tries < SendRetryCount {
-			req.Tries++
-			atomic.AddInt64(&(srvStats.RetryCount), 1)
-			logf["resend_cnt"] = req.Tries
-
-			select {
-			case retryq <- req:
-				LogWithFields(logf).
-					Debugf("Retry to enqueue into retryq because of http connection error with FCM.")
-			default:
-				LogWithFields(logf).
-					Warnf("Supervisor retry queue is full.")
-			}
-		} else {
-			LogWithFields(logf).
-				Warnf("Retry count is over than %d. Could not deliver notification.", SendRetryCount)
-		}
+		retry(retryq, req, resp.Err, logf)
 		return
 	}
 
@@ -395,12 +382,17 @@ func handleFCMResponse(resp SenderResponse, retryq chan<- Request, cmdq chan Com
 			LogWithFields(logf).Info("Succeeded to send a notification")
 			continue
 		}
-		// handle error response each registration_id
-		atomic.AddInt64(&(srvStats.ErrCount), 1)
 		switch err.Error() {
+		case fcmv1.Internal, fcmv1.Unavailable:
+			LogWithFields(logf).Warn("retrying:", err)
+			retry(retryq, resp.Req, err, logf)
+		case fcmv1.QuotaExceeded:
+			LogWithFields(logf).Warn("retrying after 1 min:", err)
+			time.AfterFunc(time.Minute, func() { retry(retryq, resp.Req, err, logf) })
 		case fcmv1.Unregistered, fcmv1.InvalidArgument, fcmv1.NotFound:
+			LogWithFields(logf).Errorf("calling error hook: %s", err)
+			atomic.AddInt64(&(srvStats.ErrCount), 1)
 			onResponse(result, errorResponseHandler.HookCmd(), cmdq)
-			LogWithFields(logf).Errorf("%s", err)
 		default:
 			LogWithFields(logf).Errorf("Unknown error message: %s", err)
 		}
@@ -438,7 +430,7 @@ func spawnSender(wq <-chan Request, respq chan<- SenderResponse, wgrp *sync.Wait
 			no := req.Notification.(apns.Notification)
 			start := time.Now()
 			results, err := ac.Send(no)
-			respTime := time.Now().Sub(start).Seconds()
+			respTime := time.Since(start).Seconds()
 			rs := make([]Result, 0, len(results))
 			for _, v := range results {
 				rs = append(rs, v)
@@ -459,7 +451,7 @@ func spawnSender(wq <-chan Request, respq chan<- SenderResponse, wgrp *sync.Wait
 			p := req.Notification.(fcmv1.Payload)
 			start := time.Now()
 			results, err := fcv1.Send(p)
-			respTime := time.Now().Sub(start).Seconds()
+			respTime := time.Since(start).Seconds()
 			rs := make([]Result, 0, len(results))
 			for _, v := range results {
 				rs = append(rs, v)

supervisor_test.go

@@ -58,6 +58,7 @@ func (tr TestResponseHandler) HookCmd() string {
 func init() {
 	logrus.SetLevel(logrus.WarnLevel)
 	conf.Apns.Host = gunfish.MockServer
+	gunfish.RetryBackoff = false // for testing
 }
 
 func TestEnqueuRequestToSupervisor(t *testing.T) {

test/gunfish_test.toml

@@ -11,3 +11,9 @@ cert_file = "{{ env `PROJECT_ROOT` `.` }}/test/server.crt"
 key_file = "{{ env `PROJECT_ROOT` `.` }}/test/server.key"
 request_per_sec = 2000
 sender_num = 50
+
+[fcm_v1]
+# google_application_credentials = "{{ env `PROJECT_ROOT` `.` }}/credentials.json"
+enabled = true
+endpoint = "http://localhost:8888/v1/projects"
+projectid = "test"

test/tools/fcmv1mock/fcmv1mock.go

@@ -0,0 +1,29 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"log"
+	"net/http"
+
+	"github.com/kayac/Gunfish/mock"
+)
+
+func main() {
+	var (
+		port      int
+		projectID string
+		verbose   bool
+	)
+
+	flag.IntVar(&port, "port", 8888, "fcmv1 mock server port")
+	flag.StringVar(&projectID, "project-id", "test", "fcmv1 mock project id")
+	flag.BoolVar(&verbose, "verbose", false, "verbose flag")
+	flag.Parse()
+
+	mux := mock.FCMv1MockServer(projectID, verbose)
+	log.Println("start fcmv1mock server port:", port, "project_id:", projectID)
+	if err := http.ListenAndServe(fmt.Sprintf(":%d", port), mux); err != nil {
+		log.Fatal(err)
+	}
+}