@@ -6,6 +6,7 @@ require ( | |||||
github.com/go-redis/redis/v8 v8.11.4 | github.com/go-redis/redis/v8 v8.11.4 | ||||
github.com/gorilla/mux v1.8.0 | github.com/gorilla/mux v1.8.0 | ||||
github.com/prometheus/client_golang v1.12.1 | github.com/prometheus/client_golang v1.12.1 | ||||
github.com/tevino/abool/v2 v2.0.1 | |||||
) | ) | ||||
require ( | require ( | ||||
@@ -208,6 +208,8 @@ github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXf | |||||
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= | github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= | ||||
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= | github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= | ||||
github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA= | github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA= | ||||
github.com/tevino/abool/v2 v2.0.1 h1:OF7FC5V5z3yAWyixbc32ecEzrgAJCsPkVOsPM2qoZPI= | |||||
github.com/tevino/abool/v2 v2.0.1/go.mod h1:+Lmlqk6bHDWHqN1cbxqhwEAwMPXgc8I1SDEamtseuXY= | |||||
github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= | github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= | ||||
github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= | github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= | ||||
github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= | github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= | ||||
@@ -20,6 +20,7 @@ import ( | |||||
"github.com/go-redis/redis/v8" | "github.com/go-redis/redis/v8" | ||||
"github.com/gorilla/mux" | "github.com/gorilla/mux" | ||||
"github.com/prometheus/client_golang/prometheus/promhttp" | "github.com/prometheus/client_golang/prometheus/promhttp" | ||||
"github.com/tevino/abool/v2" | |||||
) | ) | ||||
const ( | const ( | ||||
@@ -220,6 +221,7 @@ type GlobalBackfeedManager struct { | |||||
BackfeedRedis *redis.ClusterClient | BackfeedRedis *redis.ClusterClient | ||||
LegacyRedis *redis.Client | LegacyRedis *redis.Client | ||||
Lock sync.RWMutex | Lock sync.RWMutex | ||||
Populated *abool.AtomicBool | |||||
} | } | ||||
func (that *GlobalBackfeedManager) RefreshFeeds() error { | func (that *GlobalBackfeedManager) RefreshFeeds() error { | ||||
@@ -330,6 +332,9 @@ func (that *GlobalBackfeedManager) RefreshFeeds() error { | |||||
<-projectBackfeedManager.Done | <-projectBackfeedManager.Done | ||||
log.Printf("removed project: %s", project) | log.Printf("removed project: %s", project) | ||||
} | } | ||||
if !that.Populated.IsSet() { | |||||
that.Populated.Set() | |||||
} | |||||
return nil | return nil | ||||
} | } | ||||
@@ -394,7 +399,7 @@ func (that *GlobalBackfeedManager) GetFeed(slug string) *ProjectBackfeedManager | |||||
return projectBackfeedManager | return projectBackfeedManager | ||||
} | } | ||||
func (that *GlobalBackfeedManager) Handle(res http.ResponseWriter, req *http.Request) { | |||||
func (that *GlobalBackfeedManager) HandleLegacy(res http.ResponseWriter, req *http.Request) { | |||||
defer req.Body.Close() | defer req.Body.Close() | ||||
vars := mux.Vars(req) | vars := mux.Vars(req) | ||||
@@ -454,7 +459,32 @@ func (that *GlobalBackfeedManager) Handle(res http.ResponseWriter, req *http.Req | |||||
return | return | ||||
} | } | ||||
func (that *GlobalBackfeedManager) HandleHealth(res http.ResponseWriter, req *http.Request) { | |||||
if that.Populated.IsNotSet() { | |||||
WriteResponse(res, http.StatusServiceUnavailable, fmt.Errorf("%s", "backfeed not populated")) | |||||
return | |||||
} | |||||
err := that.LegacyRedis.Ping(req.Context()).Err() | |||||
if err != nil { | |||||
WriteResponse(res, http.StatusInternalServerError, fmt.Errorf("failed to ping legacy redis: %s", err)) | |||||
return | |||||
} | |||||
err = that.BackfeedRedis.ForEachShard(req.Context(), func(ctx context.Context, client *redis.Client) error { | |||||
return client.Ping(ctx).Err() | |||||
}) | |||||
if err != nil { | |||||
WriteResponse(res, http.StatusInternalServerError, fmt.Errorf("failed to ping backfeed redis: %s", err)) | |||||
return | |||||
} | |||||
WriteResponse(res, http.StatusOK, "ok") | |||||
} | |||||
func (that *GlobalBackfeedManager) HandlePing(res http.ResponseWriter, _ *http.Request) { | |||||
WriteResponse(res, http.StatusOK, "pong") | |||||
} | |||||
func (that *GlobalBackfeedManager) CancelAllFeeds() { | func (that *GlobalBackfeedManager) CancelAllFeeds() { | ||||
that.Populated.UnSet() | |||||
that.Cancel() | that.Cancel() | ||||
for project, projectBackfeedManager := range that.ActiveFeeds { | for project, projectBackfeedManager := range that.ActiveFeeds { | ||||
log.Printf("waiting for %s channel to shut down...", project) | log.Printf("waiting for %s channel to shut down...", project) | ||||
@@ -513,7 +543,9 @@ func main() { | |||||
log.Panicf("unable to set up backfeed projects: %s", err) | log.Panicf("unable to set up backfeed projects: %s", err) | ||||
} | } | ||||
r := mux.NewRouter() | r := mux.NewRouter() | ||||
r.Methods(http.MethodPost).Path("/legacy/{slug}").HandlerFunc(globalBackfeedManager.Handle) | |||||
r.Methods(http.MethodPost).Path("/legacy/{slug}").HandlerFunc(globalBackfeedManager.HandleLegacy) | |||||
r.Methods(http.MethodGet).Path("/ping").HandlerFunc(globalBackfeedManager.HandlePing) | |||||
r.Methods(http.MethodGet).Path("/health").HandlerFunc(globalBackfeedManager.HandleHealth) | |||||
rMetrics := mux.NewRouter() | rMetrics := mux.NewRouter() | ||||
rMetrics.PathPrefix("/debug/pprof/").Handler(http.DefaultServeMux) | rMetrics.PathPrefix("/debug/pprof/").Handler(http.DefaultServeMux) | ||||
rMetrics.Path("/metrics").Handler(promhttp.Handler()) | rMetrics.Path("/metrics").Handler(promhttp.Handler()) | ||||
@@ -0,0 +1,21 @@ | |||||
The MIT License (MIT) | |||||
Copyright (c) 2016 Tevin Zhang | |||||
Permission is hereby granted, free of charge, to any person obtaining a copy | |||||
of this software and associated documentation files (the "Software"), to deal | |||||
in the Software without restriction, including without limitation the rights | |||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell | |||||
copies of the Software, and to permit persons to whom the Software is | |||||
furnished to do so, subject to the following conditions: | |||||
The above copyright notice and this permission notice shall be included in all | |||||
copies or substantial portions of the Software. | |||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR | |||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, | |||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE | |||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER | |||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, | |||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE | |||||
SOFTWARE. |
@@ -0,0 +1,87 @@ | |||||
// Package abool provides atomic Boolean type for cleaner code and | |||||
// better performance. | |||||
package abool | |||||
import ( | |||||
"encoding/json" | |||||
"sync/atomic" | |||||
) | |||||
// New creates an AtomicBool with default set to false. | |||||
func New() *AtomicBool { | |||||
return new(AtomicBool) | |||||
} | |||||
// NewBool creates an AtomicBool with given default value. | |||||
func NewBool(ok bool) *AtomicBool { | |||||
ab := New() | |||||
if ok { | |||||
ab.Set() | |||||
} | |||||
return ab | |||||
} | |||||
// AtomicBool is an atomic Boolean. | |||||
// Its methods are all atomic, thus safe to be called by multiple goroutines simultaneously. | |||||
// Note: When embedding into a struct one should always use *AtomicBool to avoid copy. | |||||
type AtomicBool int32 | |||||
// Set sets the Boolean to true. | |||||
func (ab *AtomicBool) Set() { | |||||
atomic.StoreInt32((*int32)(ab), 1) | |||||
} | |||||
// UnSet sets the Boolean to false. | |||||
func (ab *AtomicBool) UnSet() { | |||||
atomic.StoreInt32((*int32)(ab), 0) | |||||
} | |||||
// IsSet returns whether the Boolean is true. | |||||
func (ab *AtomicBool) IsSet() bool { | |||||
return atomic.LoadInt32((*int32)(ab)) == 1 | |||||
} | |||||
// IsNotSet returns whether the Boolean is false. | |||||
func (ab *AtomicBool) IsNotSet() bool { | |||||
return !ab.IsSet() | |||||
} | |||||
// SetTo sets the boolean with given Boolean. | |||||
func (ab *AtomicBool) SetTo(yes bool) { | |||||
if yes { | |||||
atomic.StoreInt32((*int32)(ab), 1) | |||||
} else { | |||||
atomic.StoreInt32((*int32)(ab), 0) | |||||
} | |||||
} | |||||
// SetToIf sets the Boolean to new only if the Boolean matches the old. | |||||
// Returns whether the set was done. | |||||
func (ab *AtomicBool) SetToIf(old, new bool) (set bool) { | |||||
var o, n int32 | |||||
if old { | |||||
o = 1 | |||||
} | |||||
if new { | |||||
n = 1 | |||||
} | |||||
return atomic.CompareAndSwapInt32((*int32)(ab), o, n) | |||||
} | |||||
// MarshalJSON behaves the same as if the AtomicBool is a builtin.bool. | |||||
// NOTE: There's no lock during the process, usually it shouldn't be called with other methods in parallel. | |||||
func (ab *AtomicBool) MarshalJSON() ([]byte, error) { | |||||
return json.Marshal(ab.IsSet()) | |||||
} | |||||
// UnmarshalJSON behaves the same as if the AtomicBool is a builtin.bool. | |||||
// NOTE: There's no lock during the process, usually it shouldn't be called with other methods in parallel. | |||||
func (ab *AtomicBool) UnmarshalJSON(b []byte) error { | |||||
var v bool | |||||
err := json.Unmarshal(b, &v) | |||||
if err == nil { | |||||
ab.SetTo(v) | |||||
} | |||||
return err | |||||
} |
@@ -48,6 +48,9 @@ github.com/prometheus/common/model | |||||
github.com/prometheus/procfs | github.com/prometheus/procfs | ||||
github.com/prometheus/procfs/internal/fs | github.com/prometheus/procfs/internal/fs | ||||
github.com/prometheus/procfs/internal/util | github.com/prometheus/procfs/internal/util | ||||
# github.com/tevino/abool/v2 v2.0.1 | |||||
## explicit; go 1.17 | |||||
github.com/tevino/abool/v2 | |||||
# golang.org/x/sys v0.0.0-20220114195835-da31bd327af9 | # golang.org/x/sys v0.0.0-20220114195835-da31bd327af9 | ||||
## explicit; go 1.17 | ## explicit; go 1.17 | ||||
golang.org/x/sys/internal/unsafeheader | golang.org/x/sys/internal/unsafeheader | ||||