From 47fcb52354377ea1f0fbfbd21d66cef6870e89d4 Mon Sep 17 00:00:00 2001 From: Jonathan Lange Date: Tue, 14 Jun 2016 10:42:20 +0100 Subject: [PATCH] Optional memcached between probes and S3 If given settings for memcached, services will store & fetch reports from memcache after checking their in-process cache but before fetching from S3. --- app/multitenant/common.go | 20 +- app/multitenant/dynamo_collector.go | 166 +++-- app/multitenant/memcache_client.go | 170 +++++ prog/app.go | 9 +- prog/main.go | 18 +- .../bradfitz/gomemcache/memcache/LICENSE | 202 ++++++ .../bradfitz/gomemcache/memcache/memcache.go | 666 ++++++++++++++++++ .../bradfitz/gomemcache/memcache/selector.go | 114 +++ vendor/manifest | 9 + 9 files changed, 1319 insertions(+), 55 deletions(-) create mode 100644 app/multitenant/memcache_client.go create mode 100644 vendor/github.com/bradfitz/gomemcache/memcache/LICENSE create mode 100644 vendor/github.com/bradfitz/gomemcache/memcache/memcache.go create mode 100644 vendor/github.com/bradfitz/gomemcache/memcache/selector.go diff --git a/app/multitenant/common.go b/app/multitenant/common.go index 4d0bee7d6..564afe4a0 100644 --- a/app/multitenant/common.go +++ b/app/multitenant/common.go @@ -13,10 +13,28 @@ func errorCode(err error) string { return "500" } +// timeRequest runs 'f' and records how long it took in the given Prometheus +// metric. If 'f' returns successfully, record a "200". Otherwise, record +// "500". +// +// If you want more complicated logic for translating errors into statuses, +// use 'timeRequestStatus'. func timeRequest(method string, metric *prometheus.SummaryVec, f func() error) error { + return timeRequestStatus(method, metric, errorCode, f) +} + +// timeRequestStatus runs 'f' and records how long it took in the given +// Prometheus metric. +// +// toStatusCode is a function that translates errors returned by 'f' into +// HTTP-like status codes. +func timeRequestStatus(method string, metric *prometheus.SummaryVec, toStatusCode func(error) string, f func() error) error { + if toStatusCode == nil { + toStatusCode = errorCode + } startTime := time.Now() err := f() duration := time.Now().Sub(startTime) - metric.WithLabelValues(method, errorCode(err)).Observe(float64(duration.Nanoseconds())) + metric.WithLabelValues(method, toStatusCode(err)).Observe(float64(duration.Nanoseconds())) return err } diff --git a/app/multitenant/dynamo_collector.go b/app/multitenant/dynamo_collector.go index 0d5181d14..b94621b2a 100644 --- a/app/multitenant/dynamo_collector.go +++ b/app/multitenant/dynamo_collector.go @@ -15,6 +15,7 @@ import ( "github.com/aws/aws-sdk-go/service/dynamodb" "github.com/aws/aws-sdk-go/service/s3" "github.com/bluele/gcache" + "github.com/bradfitz/gomemcache/memcache" "github.com/nats-io/nats" "github.com/prometheus/client_golang/prometheus" "golang.org/x/net/context" @@ -24,12 +25,16 @@ import ( ) const ( - hourField = "hour" - tsField = "ts" - reportField = "report" - reportCacheSize = (15 / 3) * 10 * 5 // (window size * report rate) * number of hosts per user * number of users - reportCacheExpiration = 15 * time.Second - natsTimeout = 10 * time.Second + hourField = "hour" + tsField = "ts" + reportField = "report" + reportCacheSize = (15 / 3) * 10 * 5 // (window size * report rate) * number of hosts per user * number of users + reportCacheExpiration = 15 * time.Second + memcacheExpiration = 15 // seconds + memcacheUpdateInterval = 1 * time.Minute + natsTimeout = 10 * time.Second + hitLabel = "hit" + missLabel = "miss" ) var ( @@ -38,16 +43,6 @@ var ( Name: "dynamo_request_duration_nanoseconds", Help: "Time spent doing DynamoDB requests.", }, []string{"method", "status_code"}) - dynamoCacheHits = prometheus.NewCounter(prometheus.CounterOpts{ - Namespace: "scope", - Name: "dynamo_cache_hits", - Help: "Reports fetches that hit local cache.", - }) - dynamoCacheMiss = prometheus.NewCounter(prometheus.CounterOpts{ - Namespace: "scope", - Name: "dynamo_cache_miss", - Help: "Reports fetches that miss local cache.", - }) dynamoConsumedCapacity = prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: "scope", Name: "dynamo_consumed_capacity", @@ -59,6 +54,12 @@ var ( Help: "Size of data read / written from dynamodb.", }, []string{"method"}) + inProcessCacheCounter = prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "scope", + Name: "in_process_cache", + Help: "Reports fetches that hit in-process cache.", + }, []string{"result"}) + reportSize = prometheus.NewCounter(prometheus.CounterOpts{ Namespace: "scope", Name: "report_size_bytes", @@ -80,10 +81,9 @@ var ( func init() { prometheus.MustRegister(dynamoRequestDuration) - prometheus.MustRegister(dynamoCacheHits) - prometheus.MustRegister(dynamoCacheMiss) prometheus.MustRegister(dynamoConsumedCapacity) prometheus.MustRegister(dynamoValueSize) + prometheus.MustRegister(inProcessCacheCounter) prometheus.MustRegister(reportSize) prometheus.MustRegister(s3RequestDuration) prometheus.MustRegister(natsRequests) @@ -95,6 +95,11 @@ type DynamoDBCollector interface { CreateTables() error } +// ReportStore is a thing that we can get reports from. +type ReportStore interface { + FetchReports([]string) ([]report.Report, []string, error) +} + type dynamoDBCollector struct { userIDer UserIDer db *dynamodb.DynamoDB @@ -102,7 +107,8 @@ type dynamoDBCollector struct { tableName string bucketName string merger app.Merger - cache gcache.Cache + inProcess inProcessStore + memcache *MemcacheClient nats *nats.Conn waitersLock sync.Mutex @@ -128,7 +134,8 @@ type watchKey struct { func NewDynamoDBCollector( userIDer UserIDer, dynamoDBConfig, s3Config *aws.Config, - tableName, bucketName, natsHost string, + tableName, bucketName, natsHost, memcachedHost string, + memcachedTimeout time.Duration, memcachedService string, ) (DynamoDBCollector, error) { var nc *nats.Conn if natsHost != "" { @@ -139,6 +146,22 @@ func NewDynamoDBCollector( } } + var memcacheClient *MemcacheClient + if memcachedHost != "" { + var err error + memcacheClient, err = NewMemcacheClient(memcachedHost, memcachedTimeout, memcachedService, memcacheUpdateInterval, memcacheExpiration) + if err != nil { + // TODO(jml): Ideally, we wouldn't abort here, we would instead + // log errors when we try to use the memcache & fail to do so, as + // aborting here introduces ordering dependencies into our + // deployment. + // + // Note: this error only happens when either the memcachedHost or + // any of the SRV records that it points to fail to resolve. + return nil, err + } + } + return &dynamoDBCollector{ db: dynamodb.New(session.New(dynamoDBConfig)), s3: s3.New(session.New(s3Config)), @@ -146,7 +169,8 @@ func NewDynamoDBCollector( tableName: tableName, bucketName: bucketName, merger: app.NewSmartMerger(), - cache: gcache.New(reportCacheSize).LRU().Expiration(reportCacheExpiration).Build(), + inProcess: newInProcessStore(reportCacheSize, reportCacheExpiration), + memcache: memcacheClient, nats: nc, waiters: map[watchKey]*nats.Subscription{}, }, nil @@ -252,20 +276,6 @@ func (c *dynamoDBCollector) getReportKeys(rowKey string, start, end time.Time) ( return result, nil } -func (c *dynamoDBCollector) getCached(reportKeys []string) ([]report.Report, []string) { - foundReports := []report.Report{} - missingReports := []string{} - for _, reportKey := range reportKeys { - rpt, err := c.cache.Get(reportKey) - if err == nil { - foundReports = append(foundReports, rpt.(report.Report)) - } else { - missingReports = append(missingReports, reportKey) - } - } - return foundReports, missingReports -} - // Fetch multiple reports in parallel from S3. func (c *dynamoDBCollector) getNonCached(reportKeys []string) ([]report.Report, error) { type result struct { @@ -291,7 +301,7 @@ func (c *dynamoDBCollector) getNonCached(reportKeys []string) ([]report.Report, return nil, r.err } reports = append(reports, *r.report) - c.cache.Set(r.key, *r.report) + c.inProcess.StoreReport(r.key, *r.report) } return reports, nil } @@ -315,24 +325,46 @@ func (c *dynamoDBCollector) getNonCachedReport(reportKey string) (*report.Report func (c *dynamoDBCollector) getReports(userid string, row int64, start, end time.Time) ([]report.Report, error) { rowKey := fmt.Sprintf("%s-%s", userid, strconv.FormatInt(row, 10)) - reportKeys, err := c.getReportKeys(rowKey, start, end) + missing, err := c.getReportKeys(rowKey, start, end) if err != nil { return nil, err } - cachedReports, missing := c.getCached(reportKeys) - dynamoCacheHits.Add(float64(len(cachedReports))) - dynamoCacheMiss.Add(float64(len(missing))) - if len(missing) == 0 { - return cachedReports, nil + var reports []report.Report + for _, store := range []ReportStore{c.inProcess, c.memcache} { + if store == nil { + continue + } + found, missing, err := store.FetchReports(missing) + if err != nil { + log.Warningf("Error fetching from cache: %v", err) + } + reports = append(reports, found...) + if len(missing) == 0 { + return reports, nil + } } - fetchedReport, err := c.getNonCached(missing) + fetchedReports, err := c.getNonCached(missing) if err != nil { return nil, err } - return append(cachedReports, fetchedReport...), nil + return append(reports, fetchedReports...), nil +} + +func memcacheStatusCode(err error) string { + // See https://godoc.org/github.com/bradfitz/gomemcache/memcache#pkg-variables + switch err { + case nil: + return "200" + case memcache.ErrCacheMiss: + return "404" + case memcache.ErrMalformedKey: + return "400" + default: + return "500" + } } func (c *dynamoDBCollector) Report(ctx context.Context) (report.Report, error) { @@ -402,7 +434,20 @@ func (c *dynamoDBCollector) Add(ctx context.Context, rep report.Report) error { return err } - // third, put the key in dynamodb + // third, put it in memcache + if c.memcache != nil { + err = timeRequestStatus("Put", memcacheRequestDuration, memcacheStatusCode, func() error { + return c.memcache.StoreBytes(s3Key, buf.Bytes()) + }) + if err != nil { + // NOTE: We don't abort here because failing to store in memcache + // doesn't actually break anything else -- it's just an + // optimization. + log.Warningf("Could not store %v in memcache: %v", s3Key, err) + } + } + + // fourth, put the key in dynamodb dynamoValueSize.WithLabelValues("PutItem"). Add(float64(len(s3Key))) @@ -509,3 +554,34 @@ func (c *dynamoDBCollector) UnWait(ctx context.Context, waiter chan struct{}) { log.Errorf("Error on unsubscribe: %v", err) } } + +type inProcessStore struct { + cache gcache.Cache +} + +// newInProcessStore creates an in-process store for reports. +func newInProcessStore(size int, expiration time.Duration) inProcessStore { + return inProcessStore{gcache.New(size).LRU().Expiration(expiration).Build()} +} + +// FetchReports retrieves the given reports from the store. +func (c inProcessStore) FetchReports(keys []string) ([]report.Report, []string, error) { + found := []report.Report{} + missing := []string{} + for _, key := range keys { + rpt, err := c.cache.Get(key) + if err == nil { + found = append(found, rpt.(report.Report)) + } else { + missing = append(missing, key) + } + } + inProcessCacheCounter.WithLabelValues(hitLabel).Add(float64(len(found))) + inProcessCacheCounter.WithLabelValues(missLabel).Add(float64(len(missing))) + return found, missing, nil +} + +// StoreReport stores a report in the store. +func (c inProcessStore) StoreReport(key string, report report.Report) { + c.cache.Set(key, report) +} diff --git a/app/multitenant/memcache_client.go b/app/multitenant/memcache_client.go new file mode 100644 index 000000000..cd0a88b85 --- /dev/null +++ b/app/multitenant/memcache_client.go @@ -0,0 +1,170 @@ +package multitenant + +import ( + "bytes" + "fmt" + "net" + "sort" + "sync" + "time" + + log "github.com/Sirupsen/logrus" + "github.com/bradfitz/gomemcache/memcache" + "github.com/prometheus/client_golang/prometheus" + "github.com/weaveworks/scope/report" +) + +var ( + memcacheCounter = prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "scope", + Name: "memcache", + Help: "Reports that missed our in-memory cache but went to our memcache", + }, []string{"result"}) + + memcacheRequestDuration = prometheus.NewSummaryVec(prometheus.SummaryOpts{ + Namespace: "scope", + Name: "memcache_request_duration_nanoseconds", + Help: "Time spent doing memcache requests.", + }, []string{"method", "status_code"}) +) + +func init() { + prometheus.MustRegister(memcacheCounter) + prometheus.MustRegister(memcacheRequestDuration) +} + +// MemcacheClient is a memcache client that gets its server list from SRV +// records, and periodically updates that ServerList. +type MemcacheClient struct { + client *memcache.Client + serverList *memcache.ServerList + expiration int32 + hostname string + service string + + quit chan struct{} + wait sync.WaitGroup +} + +// NewMemcacheClient creates a new MemcacheClient that gets its server list +// from SRV and updates the server list on a regular basis. +func NewMemcacheClient(host string, timeout time.Duration, service string, updateInterval time.Duration, expiration int32) (*MemcacheClient, error) { + var servers memcache.ServerList + client := memcache.NewFromSelector(&servers) + client.Timeout = timeout + + newClient := &MemcacheClient{ + client: client, + serverList: &servers, + expiration: expiration, + hostname: host, + service: service, + quit: make(chan struct{}), + } + err := newClient.updateMemcacheServers() + if err != nil { + return nil, err + } + + newClient.wait.Add(1) + go newClient.updateLoop(updateInterval) + return newClient, nil +} + +// Stop the memcache client. +func (c *MemcacheClient) Stop() { + close(c.quit) + c.wait.Wait() +} + +func (c *MemcacheClient) updateLoop(updateInterval time.Duration) error { + defer c.wait.Done() + ticker := time.NewTicker(updateInterval) + var err error + for { + select { + case <-ticker.C: + err = c.updateMemcacheServers() + if err != nil { + log.Warningf("Error updating memcache servers: %v", err) + } + case <-c.quit: + ticker.Stop() + } + } +} + +// updateMemcacheServers sets a memcache server list from SRV records. SRV +// priority & weight are ignored. +func (c *MemcacheClient) updateMemcacheServers() error { + _, addrs, err := net.LookupSRV(c.service, "tcp", c.hostname) + if err != nil { + return err + } + var servers []string + for _, srv := range addrs { + servers = append(servers, fmt.Sprintf("%s:%d", srv.Target, srv.Port)) + } + // ServerList deterministically maps keys to _index_ of the server list. + // Since DNS returns records in different order each time, we sort to + // guarantee best possible match between nodes. + sort.Strings(servers) + return c.serverList.SetServers(servers...) +} + +// FetchReports gets reports from memcache. +func (c *MemcacheClient) FetchReports(keys []string) ([]report.Report, []string, error) { + var found map[string]*memcache.Item + err := timeRequestStatus("Get", memcacheRequestDuration, memcacheStatusCode, func() error { + var err error + found, err = c.client.GetMulti(keys) + return err + }) + if err != nil { + return nil, keys, err + } + + // Decode all the reports in parallel. + type result struct { + key string + report *report.Report + } + ch := make(chan result, len(keys)) + var missing []string + for _, key := range keys { + item, ok := found[key] + if !ok { + missing = append(missing, key) + continue + } + go func(key string) { + rep, err := report.MakeFromBinary(bytes.NewReader(item.Value)) + if err != nil { + log.Warningf("Corrupt report in memcache %v: %v", key, err) + ch <- result{key: key} + return + } + ch <- result{report: rep} + }(key) + } + + var reports []report.Report + for i := 0; i < len(keys)-len(missing); i++ { + r := <-ch + if r.report == nil { + missing = append(missing, r.key) + } else { + reports = append(reports, *r.report) + } + } + + memcacheCounter.WithLabelValues(hitLabel).Add(float64(len(reports))) + memcacheCounter.WithLabelValues(missLabel).Add(float64(len(missing))) + return reports, missing, nil +} + +// StoreBytes stores a report, expecting the report to be serialized already. +func (c *MemcacheClient) StoreBytes(key string, content []byte) error { + item := memcache.Item{Key: key, Value: content, Expiration: c.expiration} + return c.client.Set(&item) +} diff --git a/prog/app.go b/prog/app.go index e279d1759..536a54755 100644 --- a/prog/app.go +++ b/prog/app.go @@ -76,7 +76,7 @@ func awsConfigFromURL(url *url.URL) (*aws.Config, error) { return config, nil } -func collectorFactory(userIDer multitenant.UserIDer, collectorURL, s3URL, natsHostname string, window time.Duration, createTables bool) (app.Collector, error) { +func collectorFactory(userIDer multitenant.UserIDer, collectorURL, s3URL, natsHostname, memcachedHostname string, memcachedTimeout time.Duration, memcachedService string, window time.Duration, createTables bool) (app.Collector, error) { if collectorURL == "local" { return app.NewCollector(window), nil } @@ -102,7 +102,9 @@ func collectorFactory(userIDer multitenant.UserIDer, collectorURL, s3URL, natsHo tableName := strings.TrimPrefix(parsed.Path, "/") bucketName := strings.TrimPrefix(s3.Path, "/") dynamoCollector, err := multitenant.NewDynamoDBCollector( - userIDer, dynamoDBConfig, s3Config, tableName, bucketName, natsHostname) + userIDer, dynamoDBConfig, s3Config, tableName, bucketName, natsHostname, + memcachedHostname, memcachedTimeout, memcachedService, + ) if err != nil { return nil, err } @@ -183,7 +185,8 @@ func appMain(flags appFlags) { } collector, err := collectorFactory( - userIDer, flags.collectorURL, flags.s3URL, flags.natsHostname, flags.window, flags.awsCreateTables) + userIDer, flags.collectorURL, flags.s3URL, flags.natsHostname, flags.memcachedHostname, + flags.memcachedTimeout, flags.memcachedService, flags.window, flags.awsCreateTables) if err != nil { log.Fatalf("Error creating collector: %v", err) return diff --git a/prog/main.go b/prog/main.go index 20f20c0ed..41859f009 100644 --- a/prog/main.go +++ b/prog/main.go @@ -96,12 +96,15 @@ type appFlags struct { containerName string dockerEndpoint string - collectorURL string - s3URL string - controlRouterURL string - pipeRouterURL string - natsHostname string - userIDHeader string + collectorURL string + s3URL string + controlRouterURL string + pipeRouterURL string + natsHostname string + memcachedHostname string + memcachedTimeout time.Duration + memcachedService string + userIDHeader string awsCreateTables bool consulInf string @@ -178,6 +181,9 @@ func main() { flag.StringVar(&flags.app.controlRouterURL, "app.control.router", "local", "Control router to use (local or sqs)") flag.StringVar(&flags.app.pipeRouterURL, "app.pipe.router", "local", "Pipe router to use (local)") flag.StringVar(&flags.app.natsHostname, "app.nats", "", "Hostname for NATS service to use for shortcut reports. If empty, shortcut reporting will be disabled.") + flag.StringVar(&flags.app.memcachedHostname, "app.memcached.hostname", "", "Hostname for memcached service to use when caching reports. If empty, no memcached will be used.") + flag.DurationVar(&flags.app.memcachedTimeout, "app.memcached.timeout", 100*time.Millisecond, "Maximum time to wait before giving up on memcached requests.") + flag.StringVar(&flags.app.memcachedService, "app.memcached.service", "memcached", "SRV service used to discover memcache servers.") flag.StringVar(&flags.app.userIDHeader, "app.userid.header", "", "HTTP header to use as userid") flag.BoolVar(&flags.app.awsCreateTables, "app.aws.create.tables", false, "Create the tables in DynamoDB") diff --git a/vendor/github.com/bradfitz/gomemcache/memcache/LICENSE b/vendor/github.com/bradfitz/gomemcache/memcache/LICENSE new file mode 100644 index 000000000..d64569567 --- /dev/null +++ b/vendor/github.com/bradfitz/gomemcache/memcache/LICENSE @@ -0,0 +1,202 @@ + + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/vendor/github.com/bradfitz/gomemcache/memcache/memcache.go b/vendor/github.com/bradfitz/gomemcache/memcache/memcache.go new file mode 100644 index 000000000..7b5442d73 --- /dev/null +++ b/vendor/github.com/bradfitz/gomemcache/memcache/memcache.go @@ -0,0 +1,666 @@ +/* +Copyright 2011 Google Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package memcache provides a client for the memcached cache server. +package memcache + +import ( + "bufio" + "bytes" + "errors" + "fmt" + "io" + "io/ioutil" + "net" + + "strconv" + "strings" + "sync" + "time" +) + +// Similar to: +// http://code.google.com/appengine/docs/go/memcache/reference.html + +var ( + // ErrCacheMiss means that a Get failed because the item wasn't present. + ErrCacheMiss = errors.New("memcache: cache miss") + + // ErrCASConflict means that a CompareAndSwap call failed due to the + // cached value being modified between the Get and the CompareAndSwap. + // If the cached value was simply evicted rather than replaced, + // ErrNotStored will be returned instead. + ErrCASConflict = errors.New("memcache: compare-and-swap conflict") + + // ErrNotStored means that a conditional write operation (i.e. Add or + // CompareAndSwap) failed because the condition was not satisfied. + ErrNotStored = errors.New("memcache: item not stored") + + // ErrServer means that a server error occurred. + ErrServerError = errors.New("memcache: server error") + + // ErrNoStats means that no statistics were available. + ErrNoStats = errors.New("memcache: no statistics available") + + // ErrMalformedKey is returned when an invalid key is used. + // Keys must be at maximum 250 bytes long, ASCII, and not + // contain whitespace or control characters. + ErrMalformedKey = errors.New("malformed: key is too long or contains invalid characters") + + // ErrNoServers is returned when no servers are configured or available. + ErrNoServers = errors.New("memcache: no servers configured or available") +) + +// DefaultTimeout is the default socket read/write timeout. +const DefaultTimeout = 100 * time.Millisecond + +const ( + buffered = 8 // arbitrary buffered channel size, for readability + maxIdleConnsPerAddr = 2 // TODO(bradfitz): make this configurable? +) + +// resumableError returns true if err is only a protocol-level cache error. +// This is used to determine whether or not a server connection should +// be re-used or not. If an error occurs, by default we don't reuse the +// connection, unless it was just a cache error. +func resumableError(err error) bool { + switch err { + case ErrCacheMiss, ErrCASConflict, ErrNotStored, ErrMalformedKey: + return true + } + return false +} + +func legalKey(key string) bool { + if len(key) > 250 { + return false + } + for i := 0; i < len(key); i++ { + if key[i] <= ' ' || key[i] > 0x7e { + return false + } + } + return true +} + +var ( + crlf = []byte("\r\n") + space = []byte(" ") + resultOK = []byte("OK\r\n") + resultStored = []byte("STORED\r\n") + resultNotStored = []byte("NOT_STORED\r\n") + resultExists = []byte("EXISTS\r\n") + resultNotFound = []byte("NOT_FOUND\r\n") + resultDeleted = []byte("DELETED\r\n") + resultEnd = []byte("END\r\n") + resultOk = []byte("OK\r\n") + resultTouched = []byte("TOUCHED\r\n") + + resultClientErrorPrefix = []byte("CLIENT_ERROR ") +) + +// New returns a memcache client using the provided server(s) +// with equal weight. If a server is listed multiple times, +// it gets a proportional amount of weight. +func New(server ...string) *Client { + ss := new(ServerList) + ss.SetServers(server...) + return NewFromSelector(ss) +} + +// NewFromSelector returns a new Client using the provided ServerSelector. +func NewFromSelector(ss ServerSelector) *Client { + return &Client{selector: ss} +} + +// Client is a memcache client. +// It is safe for unlocked use by multiple concurrent goroutines. +type Client struct { + // Timeout specifies the socket read/write timeout. + // If zero, DefaultTimeout is used. + Timeout time.Duration + + selector ServerSelector + + lk sync.Mutex + freeconn map[string][]*conn +} + +// Item is an item to be got or stored in a memcached server. +type Item struct { + // Key is the Item's key (250 bytes maximum). + Key string + + // Value is the Item's value. + Value []byte + + // Flags are server-opaque flags whose semantics are entirely + // up to the app. + Flags uint32 + + // Expiration is the cache expiration time, in seconds: either a relative + // time from now (up to 1 month), or an absolute Unix epoch time. + // Zero means the Item has no expiration time. + Expiration int32 + + // Compare and swap ID. + casid uint64 +} + +// conn is a connection to a server. +type conn struct { + nc net.Conn + rw *bufio.ReadWriter + addr net.Addr + c *Client +} + +// release returns this connection back to the client's free pool +func (cn *conn) release() { + cn.c.putFreeConn(cn.addr, cn) +} + +func (cn *conn) extendDeadline() { + cn.nc.SetDeadline(time.Now().Add(cn.c.netTimeout())) +} + +// condRelease releases this connection if the error pointed to by err +// is nil (not an error) or is only a protocol level error (e.g. a +// cache miss). The purpose is to not recycle TCP connections that +// are bad. +func (cn *conn) condRelease(err *error) { + if *err == nil || resumableError(*err) { + cn.release() + } else { + cn.nc.Close() + } +} + +func (c *Client) putFreeConn(addr net.Addr, cn *conn) { + c.lk.Lock() + defer c.lk.Unlock() + if c.freeconn == nil { + c.freeconn = make(map[string][]*conn) + } + freelist := c.freeconn[addr.String()] + if len(freelist) >= maxIdleConnsPerAddr { + cn.nc.Close() + return + } + c.freeconn[addr.String()] = append(freelist, cn) +} + +func (c *Client) getFreeConn(addr net.Addr) (cn *conn, ok bool) { + c.lk.Lock() + defer c.lk.Unlock() + if c.freeconn == nil { + return nil, false + } + freelist, ok := c.freeconn[addr.String()] + if !ok || len(freelist) == 0 { + return nil, false + } + cn = freelist[len(freelist)-1] + c.freeconn[addr.String()] = freelist[:len(freelist)-1] + return cn, true +} + +func (c *Client) netTimeout() time.Duration { + if c.Timeout != 0 { + return c.Timeout + } + return DefaultTimeout +} + +// ConnectTimeoutError is the error type used when it takes +// too long to connect to the desired host. This level of +// detail can generally be ignored. +type ConnectTimeoutError struct { + Addr net.Addr +} + +func (cte *ConnectTimeoutError) Error() string { + return "memcache: connect timeout to " + cte.Addr.String() +} + +func (c *Client) dial(addr net.Addr) (net.Conn, error) { + type connError struct { + cn net.Conn + err error + } + + nc, err := net.DialTimeout(addr.Network(), addr.String(), c.netTimeout()) + if err == nil { + return nc, nil + } + + if ne, ok := err.(net.Error); ok && ne.Timeout() { + return nil, &ConnectTimeoutError{addr} + } + + return nil, err +} + +func (c *Client) getConn(addr net.Addr) (*conn, error) { + cn, ok := c.getFreeConn(addr) + if ok { + cn.extendDeadline() + return cn, nil + } + nc, err := c.dial(addr) + if err != nil { + return nil, err + } + cn = &conn{ + nc: nc, + addr: addr, + rw: bufio.NewReadWriter(bufio.NewReader(nc), bufio.NewWriter(nc)), + c: c, + } + cn.extendDeadline() + return cn, nil +} + +func (c *Client) onItem(item *Item, fn func(*Client, *bufio.ReadWriter, *Item) error) error { + addr, err := c.selector.PickServer(item.Key) + if err != nil { + return err + } + cn, err := c.getConn(addr) + if err != nil { + return err + } + defer cn.condRelease(&err) + if err = fn(c, cn.rw, item); err != nil { + return err + } + return nil +} + +func (c *Client) FlushAll() error { + return c.selector.Each(c.flushAllFromAddr) +} + +// Get gets the item for the given key. ErrCacheMiss is returned for a +// memcache cache miss. The key must be at most 250 bytes in length. +func (c *Client) Get(key string) (item *Item, err error) { + err = c.withKeyAddr(key, func(addr net.Addr) error { + return c.getFromAddr(addr, []string{key}, func(it *Item) { item = it }) + }) + if err == nil && item == nil { + err = ErrCacheMiss + } + return +} + +// Touch updates the expiry for the given key. The seconds parameter is either +// a Unix timestamp or, if seconds is less than 1 month, the number of seconds +// into the future at which time the item will expire. ErrCacheMiss is returned if the +// key is not in the cache. The key must be at most 250 bytes in length. +func (c *Client) Touch(key string, seconds int32) (err error) { + return c.withKeyAddr(key, func(addr net.Addr) error { + return c.touchFromAddr(addr, []string{key}, seconds) + }) +} + +func (c *Client) withKeyAddr(key string, fn func(net.Addr) error) (err error) { + if !legalKey(key) { + return ErrMalformedKey + } + addr, err := c.selector.PickServer(key) + if err != nil { + return err + } + return fn(addr) +} + +func (c *Client) withAddrRw(addr net.Addr, fn func(*bufio.ReadWriter) error) (err error) { + cn, err := c.getConn(addr) + if err != nil { + return err + } + defer cn.condRelease(&err) + return fn(cn.rw) +} + +func (c *Client) withKeyRw(key string, fn func(*bufio.ReadWriter) error) error { + return c.withKeyAddr(key, func(addr net.Addr) error { + return c.withAddrRw(addr, fn) + }) +} + +func (c *Client) getFromAddr(addr net.Addr, keys []string, cb func(*Item)) error { + return c.withAddrRw(addr, func(rw *bufio.ReadWriter) error { + if _, err := fmt.Fprintf(rw, "gets %s\r\n", strings.Join(keys, " ")); err != nil { + return err + } + if err := rw.Flush(); err != nil { + return err + } + if err := parseGetResponse(rw.Reader, cb); err != nil { + return err + } + return nil + }) +} + +// flushAllFromAddr send the flush_all command to the given addr +func (c *Client) flushAllFromAddr(addr net.Addr) error { + return c.withAddrRw(addr, func(rw *bufio.ReadWriter) error { + if _, err := fmt.Fprintf(rw, "flush_all\r\n"); err != nil { + return err + } + if err := rw.Flush(); err != nil { + return err + } + line, err := rw.ReadSlice('\n') + if err != nil { + return err + } + switch { + case bytes.Equal(line, resultOk): + break + default: + return fmt.Errorf("memcache: unexpected response line from flush_all: %q", string(line)) + } + return nil + }) +} + +func (c *Client) touchFromAddr(addr net.Addr, keys []string, expiration int32) error { + return c.withAddrRw(addr, func(rw *bufio.ReadWriter) error { + for _, key := range keys { + if _, err := fmt.Fprintf(rw, "touch %s %d\r\n", key, expiration); err != nil { + return err + } + if err := rw.Flush(); err != nil { + return err + } + line, err := rw.ReadSlice('\n') + if err != nil { + return err + } + switch { + case bytes.Equal(line, resultTouched): + break + case bytes.Equal(line, resultNotFound): + return ErrCacheMiss + default: + return fmt.Errorf("memcache: unexpected response line from touch: %q", string(line)) + } + } + return nil + }) +} + +// GetMulti is a batch version of Get. The returned map from keys to +// items may have fewer elements than the input slice, due to memcache +// cache misses. Each key must be at most 250 bytes in length. +// If no error is returned, the returned map will also be non-nil. +func (c *Client) GetMulti(keys []string) (map[string]*Item, error) { + var lk sync.Mutex + m := make(map[string]*Item) + addItemToMap := func(it *Item) { + lk.Lock() + defer lk.Unlock() + m[it.Key] = it + } + + keyMap := make(map[net.Addr][]string) + for _, key := range keys { + if !legalKey(key) { + return nil, ErrMalformedKey + } + addr, err := c.selector.PickServer(key) + if err != nil { + return nil, err + } + keyMap[addr] = append(keyMap[addr], key) + } + + ch := make(chan error, buffered) + for addr, keys := range keyMap { + go func(addr net.Addr, keys []string) { + ch <- c.getFromAddr(addr, keys, addItemToMap) + }(addr, keys) + } + + var err error + for _ = range keyMap { + if ge := <-ch; ge != nil { + err = ge + } + } + return m, err +} + +// parseGetResponse reads a GET response from r and calls cb for each +// read and allocated Item +func parseGetResponse(r *bufio.Reader, cb func(*Item)) error { + for { + line, err := r.ReadSlice('\n') + if err != nil { + return err + } + if bytes.Equal(line, resultEnd) { + return nil + } + it := new(Item) + size, err := scanGetResponseLine(line, it) + if err != nil { + return err + } + it.Value, err = ioutil.ReadAll(io.LimitReader(r, int64(size)+2)) + if err != nil { + return err + } + if !bytes.HasSuffix(it.Value, crlf) { + return fmt.Errorf("memcache: corrupt get result read") + } + it.Value = it.Value[:size] + cb(it) + } +} + +// scanGetResponseLine populates it and returns the declared size of the item. +// It does not read the bytes of the item. +func scanGetResponseLine(line []byte, it *Item) (size int, err error) { + pattern := "VALUE %s %d %d %d\r\n" + dest := []interface{}{&it.Key, &it.Flags, &size, &it.casid} + if bytes.Count(line, space) == 3 { + pattern = "VALUE %s %d %d\r\n" + dest = dest[:3] + } + n, err := fmt.Sscanf(string(line), pattern, dest...) + if err != nil || n != len(dest) { + return -1, fmt.Errorf("memcache: unexpected line in get response: %q", line) + } + return size, nil +} + +// Set writes the given item, unconditionally. +func (c *Client) Set(item *Item) error { + return c.onItem(item, (*Client).set) +} + +func (c *Client) set(rw *bufio.ReadWriter, item *Item) error { + return c.populateOne(rw, "set", item) +} + +// Add writes the given item, if no value already exists for its +// key. ErrNotStored is returned if that condition is not met. +func (c *Client) Add(item *Item) error { + return c.onItem(item, (*Client).add) +} + +func (c *Client) add(rw *bufio.ReadWriter, item *Item) error { + return c.populateOne(rw, "add", item) +} + +// Replace writes the given item, but only if the server *does* +// already hold data for this key +func (c *Client) Replace(item *Item) error { + return c.onItem(item, (*Client).replace) +} + +func (c *Client) replace(rw *bufio.ReadWriter, item *Item) error { + return c.populateOne(rw, "replace", item) +} + +// CompareAndSwap writes the given item that was previously returned +// by Get, if the value was neither modified or evicted between the +// Get and the CompareAndSwap calls. The item's Key should not change +// between calls but all other item fields may differ. ErrCASConflict +// is returned if the value was modified in between the +// calls. ErrNotStored is returned if the value was evicted in between +// the calls. +func (c *Client) CompareAndSwap(item *Item) error { + return c.onItem(item, (*Client).cas) +} + +func (c *Client) cas(rw *bufio.ReadWriter, item *Item) error { + return c.populateOne(rw, "cas", item) +} + +func (c *Client) populateOne(rw *bufio.ReadWriter, verb string, item *Item) error { + if !legalKey(item.Key) { + return ErrMalformedKey + } + var err error + if verb == "cas" { + _, err = fmt.Fprintf(rw, "%s %s %d %d %d %d\r\n", + verb, item.Key, item.Flags, item.Expiration, len(item.Value), item.casid) + } else { + _, err = fmt.Fprintf(rw, "%s %s %d %d %d\r\n", + verb, item.Key, item.Flags, item.Expiration, len(item.Value)) + } + if err != nil { + return err + } + if _, err = rw.Write(item.Value); err != nil { + return err + } + if _, err := rw.Write(crlf); err != nil { + return err + } + if err := rw.Flush(); err != nil { + return err + } + line, err := rw.ReadSlice('\n') + if err != nil { + return err + } + switch { + case bytes.Equal(line, resultStored): + return nil + case bytes.Equal(line, resultNotStored): + return ErrNotStored + case bytes.Equal(line, resultExists): + return ErrCASConflict + case bytes.Equal(line, resultNotFound): + return ErrCacheMiss + } + return fmt.Errorf("memcache: unexpected response line from %q: %q", verb, string(line)) +} + +func writeReadLine(rw *bufio.ReadWriter, format string, args ...interface{}) ([]byte, error) { + _, err := fmt.Fprintf(rw, format, args...) + if err != nil { + return nil, err + } + if err := rw.Flush(); err != nil { + return nil, err + } + line, err := rw.ReadSlice('\n') + return line, err +} + +func writeExpectf(rw *bufio.ReadWriter, expect []byte, format string, args ...interface{}) error { + line, err := writeReadLine(rw, format, args...) + if err != nil { + return err + } + switch { + case bytes.Equal(line, resultOK): + return nil + case bytes.Equal(line, expect): + return nil + case bytes.Equal(line, resultNotStored): + return ErrNotStored + case bytes.Equal(line, resultExists): + return ErrCASConflict + case bytes.Equal(line, resultNotFound): + return ErrCacheMiss + } + return fmt.Errorf("memcache: unexpected response line: %q", string(line)) +} + +// Delete deletes the item with the provided key. The error ErrCacheMiss is +// returned if the item didn't already exist in the cache. +func (c *Client) Delete(key string) error { + return c.withKeyRw(key, func(rw *bufio.ReadWriter) error { + return writeExpectf(rw, resultDeleted, "delete %s\r\n", key) + }) +} + +// DeleteAll deletes all items in the cache. +func (c *Client) DeleteAll() error { + return c.withKeyRw("", func(rw *bufio.ReadWriter) error { + return writeExpectf(rw, resultDeleted, "flush_all\r\n") + }) +} + +// Increment atomically increments key by delta. The return value is +// the new value after being incremented or an error. If the value +// didn't exist in memcached the error is ErrCacheMiss. The value in +// memcached must be an decimal number, or an error will be returned. +// On 64-bit overflow, the new value wraps around. +func (c *Client) Increment(key string, delta uint64) (newValue uint64, err error) { + return c.incrDecr("incr", key, delta) +} + +// Decrement atomically decrements key by delta. The return value is +// the new value after being decremented or an error. If the value +// didn't exist in memcached the error is ErrCacheMiss. The value in +// memcached must be an decimal number, or an error will be returned. +// On underflow, the new value is capped at zero and does not wrap +// around. +func (c *Client) Decrement(key string, delta uint64) (newValue uint64, err error) { + return c.incrDecr("decr", key, delta) +} + +func (c *Client) incrDecr(verb, key string, delta uint64) (uint64, error) { + var val uint64 + err := c.withKeyRw(key, func(rw *bufio.ReadWriter) error { + line, err := writeReadLine(rw, "%s %s %d\r\n", verb, key, delta) + if err != nil { + return err + } + switch { + case bytes.Equal(line, resultNotFound): + return ErrCacheMiss + case bytes.HasPrefix(line, resultClientErrorPrefix): + errMsg := line[len(resultClientErrorPrefix) : len(line)-2] + return errors.New("memcache: client error: " + string(errMsg)) + } + val, err = strconv.ParseUint(string(line[:len(line)-2]), 10, 64) + if err != nil { + return err + } + return nil + }) + return val, err +} diff --git a/vendor/github.com/bradfitz/gomemcache/memcache/selector.go b/vendor/github.com/bradfitz/gomemcache/memcache/selector.go new file mode 100644 index 000000000..10b04d349 --- /dev/null +++ b/vendor/github.com/bradfitz/gomemcache/memcache/selector.go @@ -0,0 +1,114 @@ +/* +Copyright 2011 Google Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package memcache + +import ( + "hash/crc32" + "net" + "strings" + "sync" +) + +// ServerSelector is the interface that selects a memcache server +// as a function of the item's key. +// +// All ServerSelector implementations must be safe for concurrent use +// by multiple goroutines. +type ServerSelector interface { + // PickServer returns the server address that a given item + // should be shared onto. + PickServer(key string) (net.Addr, error) + Each(func(net.Addr) error) error +} + +// ServerList is a simple ServerSelector. Its zero value is usable. +type ServerList struct { + mu sync.RWMutex + addrs []net.Addr +} + +// SetServers changes a ServerList's set of servers at runtime and is +// safe for concurrent use by multiple goroutines. +// +// Each server is given equal weight. A server is given more weight +// if it's listed multiple times. +// +// SetServers returns an error if any of the server names fail to +// resolve. No attempt is made to connect to the server. If any error +// is returned, no changes are made to the ServerList. +func (ss *ServerList) SetServers(servers ...string) error { + naddr := make([]net.Addr, len(servers)) + for i, server := range servers { + if strings.Contains(server, "/") { + addr, err := net.ResolveUnixAddr("unix", server) + if err != nil { + return err + } + naddr[i] = addr + } else { + tcpaddr, err := net.ResolveTCPAddr("tcp", server) + if err != nil { + return err + } + naddr[i] = tcpaddr + } + } + + ss.mu.Lock() + defer ss.mu.Unlock() + ss.addrs = naddr + return nil +} + +// Each iterates over each server calling the given function +func (ss *ServerList) Each(f func(net.Addr) error) error { + ss.mu.RLock() + defer ss.mu.RUnlock() + for _, a := range ss.addrs { + if err := f(a); nil != err { + return err + } + } + return nil +} + +// keyBufPool returns []byte buffers for use by PickServer's call to +// crc32.ChecksumIEEE to avoid allocations. (but doesn't avoid the +// copies, which at least are bounded in size and small) +var keyBufPool = sync.Pool{ + New: func() interface{} { + b := make([]byte, 256) + return &b + }, +} + +func (ss *ServerList) PickServer(key string) (net.Addr, error) { + ss.mu.RLock() + defer ss.mu.RUnlock() + if len(ss.addrs) == 0 { + return nil, ErrNoServers + } + if len(ss.addrs) == 1 { + return ss.addrs[0], nil + } + bufp := keyBufPool.Get().(*[]byte) + n := copy(*bufp, key) + cs := crc32.ChecksumIEEE((*bufp)[:n]) + keyBufPool.Put(bufp) + + return ss.addrs[cs%uint32(len(ss.addrs))], nil +} diff --git a/vendor/manifest b/vendor/manifest index 35e512c89..5b8e12c7a 100644 --- a/vendor/manifest +++ b/vendor/manifest @@ -74,6 +74,15 @@ "revision": "fb6c0b0e1ff03057a054886141927cdce6239dec", "branch": "master" }, + { + "importpath": "github.com/bradfitz/gomemcache/memcache", + "repository": "https://github.com/bradfitz/gomemcache", + "vcs": "git", + "revision": "fb1f79c6b65acda83063cbc69f6bba1522558bfc", + "branch": "master", + "path": "/memcache", + "notests": true + }, { "importpath": "github.com/c9s/goprocinfo/linux", "repository": "https://github.com/c9s/goprocinfo",