Skip to content
Snippets Groups Projects
Commit 4bedf17c authored by Dean's avatar Dean
Browse files

version 0.5.0: switch to filesystem backend

parent 3f0b9d9f
Branches
Tags
No related merge requests found
Expand all Hide whitespace changes
Inline Side-by-side
Showing
with 506 additions and 268 deletions
.gitignore
+ 3
− 1
View file @ 4bedf17c
......@@ -15,12 +15,14 @@
# Built package binary
cdn-origin
main
# Package configuration file (not including sample configuration)
cdn-origin.toml
config.toml
# MaxMind GeoLite2 Country database files
GeoLite2-Country.mmdb
# JetBrains .idea
.idea/
Dockerfile
+ 11
− 16
View file @ 4bedf17c
FROM golang:alpine
COPY cdn-origin.go $GOPATH/src/github.com/whats-this/cdn-origin/
COPY prometheus/ $GOPATH/src/github.com/whats-this/cdn-origin/prometheus/
COPY weed/ $GOPATH/src/github.com/whats-this/cdn-origin/weed/
RUN apk add --no-cache --virtual .build-deps git && \
go-wrapper download github.com/lib/pq && \
go-wrapper download github.com/prometheus/client_golang/prometheus && \
go-wrapper download github.com/Sirupsen/logrus && \
go-wrapper download github.com/spf13/pflag && \
go-wrapper download github.com/spf13/viper && \
go-wrapper download github.com/valyala/fasthttp && \
go-wrapper install github.com/whats-this/cdn-origin && \
COPY go.mod /git/owo.codes/whats-this/cdn-origin/
COPY go.sum /git/owo.codes/whats-this/cdn-origin/
COPY main.go /git/owo.codes/whats-this/cdn-origin/
COPY lib /git/owo.codes/whats-this/cdn-origin/lib
RUN apk add --no-cache --virtual .build-deps git build-base && \
cd /git/owo.codes/whats-this/cdn-origin && \
go build main.go && \
apk del .build-deps
WORKDIR $GOPATH/src/github.com/whats-this/cdn-origin
ENTRYPOINT ["go-wrapper", "run"]
WORKDIR /git/owo.codes/whats-this/cdn-origin
ENTRYPOINT ["./main"]
cdn-origin.sample.toml → config.sample.toml
+ 9
− 6
View file @ 4bedf17c
# Enable debug mode (logs requests and prints other information)
debug = false
[log]
# Log level (5=panic, 4=fatal, 3=error, 2=warn, 1=info, 0=debug)
level = 1
[database]
# PostgreSQL connection URL, see
......@@ -7,6 +8,9 @@ debug = false
# more information
connectionURL = "postgres://postgres:password@localhost/whats_this?sslmode=disable"
# Bucket to serve objects from
objectBucket = "public"
[http]
# Enable transparent response compression (only when the client Accepts it)
compressResponse = false
......@@ -49,7 +53,6 @@ debug = false
# code data will be collected. https://dev.maxmind.com/geoip/geoip2/geolite2
maxmindDBLocation = "/var/data/maxmind/GeoLite2-Country.mmdb"
[seaweed]
# SeaweedFS master URL (used for looking up volumes, volumes must be
# accessible by their publicUrl)
masterURL = "http://localhost:9333"
[files]
# Storage location of the bucket on disk
storageLocation = "/var/data/buckets/public"
go.mod 0 → 100644
+ 12
− 0
View file @ 4bedf17c
module owo.codes/whats-this/cdn-origin
require (
github.com/lib/pq v1.0.0
github.com/oschwald/maxminddb-golang v1.3.0
github.com/rs/zerolog v1.11.0
github.com/spf13/pflag v1.0.3
github.com/spf13/viper v1.3.1
github.com/valyala/fasthttp v1.2.0
golang.org/x/sys v0.0.0-20190222171317-cd391775e71e // indirect
gopkg.in/olivere/elastic.v5 v5.0.79
)
go.sum 0 → 100644
+ 62
− 0
View file @ 4bedf17c
github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
github.com/coreos/go-etcd v2.0.0+incompatible/go.mod h1:Jez6KQU2B/sWsbdaef3ED8NzMklzPG4d5KIOhIy30Tk=
github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/fortytw2/leaktest v1.2.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I=
github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
github.com/klauspost/compress v1.4.0 h1:8nsMz3tWa9SWWPL60G1V6CUsf4lLjWLTNEtibhe8gh8=
github.com/klauspost/compress v1.4.0/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
github.com/klauspost/cpuid v0.0.0-20180405133222-e7e905edc00e h1:+lIPJOWl+jSiJOc70QXJ07+2eg2Jy2EC7Mi11BWujeM=
github.com/klauspost/cpuid v0.0.0-20180405133222-e7e905edc00e/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek=
github.com/lib/pq v1.0.0 h1:X5PMW56eZitiTeO7tKzZxFCSpbFZJtkMMooicw2us9A=
github.com/lib/pq v1.0.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
github.com/magiconair/properties v1.8.0 h1:LLgXmsheXeRoUOBOjtwPQCWIYqM/LU1ayDtDePerRcY=
github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
github.com/mailru/easyjson v0.0.0-20180730094502-03f2033d19d5 h1:0x4qcEHDpruK6ML/m/YSlFUUu0UpRD3I2PHsNCuGnyA=
github.com/mailru/easyjson v0.0.0-20180730094502-03f2033d19d5/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
github.com/mitchellh/mapstructure v1.1.2 h1:fmNYVwqnSfB9mZU6OS2O6GsXM+wcskZDuKQzvN1EDeE=
github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
github.com/oschwald/maxminddb-golang v1.3.0 h1:oTh8IBSj10S5JNlUDg5WjJ1QdBMdeaZIkPEVfESSWgE=
github.com/oschwald/maxminddb-golang v1.3.0/go.mod h1:3jhIUymTJ5VREKyIhWm66LJiQt04F0UCDdodShpjWsY=
github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181zc=
github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
github.com/pkg/errors v0.8.0 h1:WdK/asTD0HN+q6hsWO3/vpuAkAr+tw6aNJNDFFf0+qw=
github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/rs/zerolog v1.11.0 h1:DRuq/S+4k52uJzBQciUcofXx45GrMC6yrEbb/CoK6+M=
github.com/rs/zerolog v1.11.0/go.mod h1:YbFCdg8HfsridGWAh22vktObvhZbQsZXe4/zB0OKkWU=
github.com/smartystreets/go-aws-auth v0.0.0-20180515143844-0c1422d1fdb9/go.mod h1:SnhjPscd9TpLiy1LpzGSKh3bXCfxxXuqd9xmQJy3slM=
github.com/spf13/afero v1.1.2 h1:m8/z1t7/fwjysjQRYbP0RD+bUIF/8tJwPdEZsI83ACI=
github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
github.com/spf13/cast v1.3.0 h1:oget//CVOEoFewqQxwr0Ej5yjygnqGkvggSE/gB35Q8=
github.com/spf13/cast v1.3.0/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
github.com/spf13/jwalterweatherman v1.0.0 h1:XHEdyB+EcvlqZamSM4ZOMGlc93t6AcsBEu9Gc1vn7yk=
github.com/spf13/jwalterweatherman v1.0.0/go.mod h1:cQK4TGJAtQXfYWX+Ddv3mKDzgVb68N+wFjFa4jdeBTo=
github.com/spf13/pflag v1.0.3 h1:zPAT6CGy6wXeQ7NtTnaTerfKOsV6V6F8agHXFiazDkg=
github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
github.com/spf13/viper v1.3.1 h1:5+8j8FTpnFV4nEImW/ofkzEt8VoOiLXxdYIDsB73T38=
github.com/spf13/viper v1.3.1/go.mod h1:ZiWeW+zYFKm7srdB9IoDzzZXaJaI5eL9QjNiN/DMA2s=
github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=
github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
github.com/valyala/fasthttp v1.2.0 h1:dzZJf2IuMiclVjdw0kkT+f9u4YdrapbNyGAN47E/qnk=
github.com/valyala/fasthttp v1.2.0/go.mod h1:4vX61m6KN+xDduDNwXrhIAVZaZaZiQ1luJk8LWSxF3s=
github.com/valyala/tcplisten v0.0.0-20161114210144-ceec8f93295a/go.mod h1:v3UYOV9WzVtRmSR+PDvWpU/qWl4Wa5LApYYX4ZtKbio=
github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q=
golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
golang.org/x/net v0.0.0-20180911220305-26e67e76b6c3/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
golang.org/x/sys v0.0.0-20181205085412-a5c9d58dba9a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190222171317-cd391775e71e h1:oF7qaQxUH6KzFdKN4ww7NpPdo53SZi4UlcksLrb2y/o=
golang.org/x/sys v0.0.0-20190222171317-cd391775e71e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/olivere/elastic.v5 v5.0.79 h1:q+FQfSQxl+xIHoEwq8RGBsb5pRB9f8rfaLh4D9jx18A=
gopkg.in/olivere/elastic.v5 v5.0.79/go.mod h1:uhHoB4o3bvX5sorxBU29rPcmBQdV2Qfg0FBrx5D6pV0=
gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
lib/db/connect.go 0 → 100644
+ 21
− 0
View file @ 4bedf17c
package db
import (
"database/sql"
// postgres driver for database/sql
_ "github.com/lib/pq"
)
// DB is the current database connection.
var DB *sql.DB
// Connect to the database using the given driver and connection URL.
func Connect(driver string, connectionURL string) error {
var err error
DB, err = sql.Open(driver, connectionURL)
if err != nil {
return err
}
return DB.Ping()
}
lib/db/models.go 0 → 100644
+ 8
− 0
View file @ 4bedf17c
package db
// Object represents a partial object from the database.
type Object struct {
ContentType *string `json:"content_type"`
DestURL *string `json:"dest_url"`
ObjectType int `json:"object_type"`
}
lib/db/queries.go 0 → 100644
+ 30
− 0
View file @ 4bedf17c
package db
import (
"database/sql"
"fmt"
)
// SelectObjectByBucketKey returns an object from a bucket and a key.
func SelectObjectByBucketKey(bucket, key string) (Object, error) {
var object Object
var contentType sql.NullString
var destURL sql.NullString
var objectType int
err := DB.QueryRow(selectObjectByBucketKey, fmt.Sprintf("%s/%s", bucket, key)).
Scan(&contentType, &destURL, &objectType)
if err != nil {
return object, err
}
// Populate object values
if contentType.Valid {
object.ContentType = &contentType.String
}
if destURL.Valid {
object.DestURL= &destURL.String
}
object.ObjectType = objectType
return object, nil
}
lib/db/sql.go 0 → 100644
+ 13
− 0
View file @ 4bedf17c
package db
var selectObjectByBucketKey = `
SELECT
content_type,
dest_url,
"type"
FROM
objects
WHERE
bucket_key = $1
LIMIT 1
`
metrics/collector.go → lib/metrics/collector.go
+ 5
− 9
View file @ 4bedf17c
......@@ -7,8 +7,8 @@ import (
"net"
"strings"
log "github.com/Sirupsen/logrus"
"github.com/oschwald/maxminddb-golang"
"github.com/rs/zerolog/log"
"gopkg.in/olivere/elastic.v5"
"gopkg.in/olivere/elastic.v5/config"
)
......@@ -95,7 +95,7 @@ func New(elasticURL string, maxmindLoc string, enableHostnameWhitelist bool, hos
return nil, fmt.Errorf("failed to parse elasticURL: %s", err)
}
if cfg.Index == "" {
log.Info(`empty index name in elasticURL, using "cdn-origin_requests"`)
log.Info().Msg(`empty index name in elasticURL, using "cdn-origin_requests"`)
cfg.Index = "cdn-origin_requests"
}
......@@ -109,11 +109,7 @@ func New(elasticURL string, maxmindLoc string, enableHostnameWhitelist bool, hos
if err != nil {
return nil, fmt.Errorf("failed to ping Elasticsearch server: %s", err)
}
// TODO: if code != 200
log.WithFields(log.Fields{
"statusCode": code,
"version": info.Version.Number,
}).Debug("elasticsearch ping success")
log.Debug().Int("statusCode", code).Str("version", info.Version.Number).Msg("elasticsearch ping success")
// Check if the index exists or create it
exists, err := client.IndexExists(cfg.Index).Do(ctx)
......@@ -121,7 +117,7 @@ func New(elasticURL string, maxmindLoc string, enableHostnameWhitelist bool, hos
return nil, fmt.Errorf("failed to determine if the index exists: %s", err)
}
if !exists {
log.WithField("index", cfg.Index).Info("creating Elasticsearch index")
log.Info().Str("index", cfg.Index).Msg("creating Elasticsearch index")
createIndex, err := client.CreateIndex(cfg.Index).
BodyString(mapping).
Do(ctx)
......@@ -141,7 +137,7 @@ func New(elasticURL string, maxmindLoc string, enableHostnameWhitelist bool, hos
return nil, fmt.Errorf("failed to determine if the timestamp pipeline exists: %s", err)
}
if len(pipelines) == 0 {
log.WithField("pipeline", "timestamp").Info("creating Elasticsearch ingest pipeline")
log.Info().Str("pipeline", "timestamp").Msg("creating Elasticsearch ingest pipeline")
putPipeline, err := elastic.NewIngestPutPipelineService(client).
Id("timestamp").
BodyString(timestampPipeline).
......
metrics/geoip.go → lib/metrics/geoip.go
+ 0
− 0
View file @ 4bedf17c
File moved
metrics/match.go → lib/metrics/match.go
+ 0
− 0
View file @ 4bedf17c
File moved
metrics/record.go → lib/metrics/record.go
+ 0
− 0
View file @ 4bedf17c
File moved
cdn-origin.go → main.go
+ 329
− 0
View file @ 4bedf17c
This diff is collapsed.
objects.sql
+ 3
− 4
View file @ 4bedf17c
......@@ -3,25 +3,24 @@ CREATE TABLE IF NOT EXISTS objects (
bucket_key VARCHAR(1088) NOT NULL UNIQUE, -- bucket + key (unique)
bucket VARCHAR(20) NOT NULL, -- uint64 bucket ID ("public" for public bucket)
"key" VARCHAR(1024) NOT NULL, -- Full bucket path to file (including directory)
random_key VARCHAR(1024), -- random key if used
dir VARCHAR(1024) NOT NULL, -- Directory of file (with trailing slash)
"type" integer NOT NULL DEFAULT 0, -- Object type enumerable (0 = file, 1 = redirect)
backend_file_id VARCHAR(33) DEFAULT NULL, -- SeaweedFS file ID (only when object.type == 0)
dest_url VARCHAR(1024) DEFAULT NULL, -- Destination URL for redirect object (only when object.type == 1)
content_type VARCHAR(255) DEFAULT 'application/octet-stream', -- Content-Type of file
content_length INT DEFAULT NULL, -- Content-Length of file
auth_hash VARCHAR(64) DEFAULT NULL, -- Authentication hash: sha256(user.id + password + object.id)
associated_user VARCHAR(36) DEFAULT NULL, -- ID of user who uploaded file
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, -- File creation timestamp
md5_hash VARCHAR(32) DEFAULT NULL -- MD5 hash of file contents (or destination URL)
);
-- Test file object: /index.md
INSERT INTO objects (bucket_key, bucket, key, dir, type, backend_file_id, content_type, content_length, md5_hash) VALUES (
INSERT INTO objects (bucket_key, bucket, key, dir, type, content_type, content_length, md5_hash) VALUES (
'public/index.txt',
'public',
'/index.txt',
'/',
0,
'1,020c3fd6ab',
'text/plain',
0,
'e2a81ac6617d7963bda5155239b4b262'
......
weed/seaweed.go deleted 100644 → 0
+ 0
− 173
View file @ 3f0b9d9f
package weed
import (
"encoding/json"
"errors"
"fmt"
"io"
"strconv"
"strings"
"time"
log "github.com/Sirupsen/logrus"
"github.com/valyala/fasthttp"
)
// lookupResponse represents a volume lookup response from the SeaweedFS master.
type lookupResponse struct {
VolumeID string `json:"volumeId"`
Locations []struct {
URL string `json:"url"`
PublicURL string `json:"publicUrl"`
} `json:"locations"`
}
// publicUrlsToSlice converts the list of locations in the lookupResponse to a slice of publicURL strings.
func (res *lookupResponse) publicURLsToSlice() []string {
var urls = []string{}
for _, loc := range res.Locations {
urls = append(urls, loc.PublicURL)
}
return urls
}
// Seaweed allows for retrieving files from a SeaweedFS cluster.
type Seaweed struct {
// Connection URI for the master.
Master string
// Volume cache to use if CacheVolumes is true.
VolumeCache *VolumeCache
// Lookup timeout for fetching volumes from master.
LookupTimeout time.Duration
}
// New creates a new instance of Seaweed.
func New(masterURI string, lookupTimeout time.Duration) *Seaweed {
return &Seaweed{
Master: masterURI,
VolumeCache: &VolumeCache{
volumeCache: map[uint32][]string{},
next: map[uint32]int{},
},
LookupTimeout: lookupTimeout,
}
}
// Get a file from a SeaweedFS cluster.
func (s *Seaweed) Get(writer io.Writer, fid string, headers map[string][]byte, query string) (int, map[string][]byte, error) {
volumeURL := s.lookupVolume(strings.Split(fid, ",")[0])
if volumeURL == "" {
return fasthttp.StatusInternalServerError, nil, errors.New("failed to retrieve volume URL")
}
requestURL := volumeURL
if !strings.HasPrefix(requestURL, "http://") && !strings.HasPrefix(requestURL, "https://") {
requestURL = "http://" + requestURL
}
if !strings.HasSuffix(requestURL, "/") {
requestURL += "/"
}
requestURL += fid
if len(query) != 0 {
requestURL += "?" + query
}
// Set request and response
req := fasthttp.AcquireRequest()
req.Reset()
req.SetRequestURI(requestURL)
if headers != nil {
for h, v := range headers {
req.Header.SetBytesV(h, v)
}
}
res := fasthttp.AcquireResponse()
defer func() {
fasthttp.ReleaseRequest(req)
fasthttp.ReleaseResponse(res)
}()
// Perform request
err := fasthttp.Do(req, res)
if err != nil {
return 0, nil, err
}
if res.StatusCode() == fasthttp.StatusOK || res.StatusCode() == fasthttp.StatusPartialContent {
if err := res.BodyWriteTo(writer); err != nil {
log.WithField("err", err).Warn("failed to set body writer for response")
return fasthttp.StatusInternalServerError, nil, err
}
}
// Get response headers
resHeaders := map[string][]byte{}
res.Header.VisitAll(func(key, value []byte) {
resHeaders[string(key)] = value
})
return res.StatusCode(), resHeaders, err
}
// Ping the master of a SeaweedFS cluster (sends a /cluster/status HTTP request to the master).
func (s *Seaweed) Ping() error {
lookupURL := fmt.Sprintf("%s/cluster/status", s.Master)
statusCode, _, err := fasthttp.GetTimeout(nil, lookupURL, s.LookupTimeout)
if err != nil {
return err
}
if statusCode != fasthttp.StatusOK {
return fmt.Errorf("expected 200 OK response status code, got %v", statusCode)
}
return nil
}
func (s *Seaweed) lookupVolume(volumeID string) string {
volumeUint64, err := strconv.ParseUint(volumeID, 10, 32)
if err != nil {
log.WithFields(log.Fields{
"err": err,
"volumeID": volumeID,
}).Warn("could not parse volume ID")
return ""
}
volumeUint32 := uint32(volumeUint64)
if uri := s.VolumeCache.GetNext(volumeUint32); uri != "" {
return uri
}
lookupURL := fmt.Sprintf("%s/dir/lookup?volumeId=%s", s.Master, volumeID)
log.WithFields(log.Fields{
"lookupURL": lookupURL,
"volumeID": volumeID,
}).Debug("looking up volume from SeaweedFS master")
statusCode, body, err := fasthttp.GetTimeout(nil, lookupURL, s.LookupTimeout)
if err != nil {
log.WithFields(log.Fields{
"err": err,
"url": lookupURL,
}).Error("failed to lookup SeaweedFS volume from master")
return ""
}
if statusCode != fasthttp.StatusOK {
log.WithFields(log.Fields{
"expected": fasthttp.StatusOK,
"got": statusCode,
}).Warn("unexpected status code while looking up SeaweedFS volume from master")
return ""
}
var res lookupResponse
err = json.Unmarshal(body, &res)
if err != nil {
log.WithFields(log.Fields{
"body": string(body),
"err": err,
}).Error("failed to parse lookup volume response from SeaweedFS master")
return ""
}
if len(res.Locations) == 0 {
log.Warn("SeaweedFS master returned no volume servers without 404ing")
return ""
}
s.VolumeCache.Add(volumeUint32, res.publicURLsToSlice())
return s.VolumeCache.GetNext(volumeUint32)
}
weed/volume-cache.go deleted 100644 → 0
+ 0
− 59
View file @ 3f0b9d9f
package weed
import "sync"
// VolumeCache stores a volume ID => volume URL map used for caching volume lookup responses from the master of a
// SeaweedFS cluster.
type VolumeCache struct {
sync.RWMutex
volumeCache map[uint32][]string
next map[uint32]int
}
// Add adds a volume ID => location URL slice mapping to the volume cache.
func (v *VolumeCache) Add(id uint32, urls []string) {
v.Lock()
defer v.Unlock()
v.volumeCache[id] = urls
v.next[id] = 0
}
// Get returns all volume server URLs for a given volume ID.
func (v *VolumeCache) Get(id uint32) []string {
v.RLock()
defer v.RUnlock()
vol, _ := v.volumeCache[id]
return vol
}
// GetNext returns the n+1th location URL for the given volume ID, n is tracked internally.
func (v *VolumeCache) GetNext(id uint32) string {
v.Lock()
defer v.Unlock()
vol, ok := v.volumeCache[id]
volLen := len(vol)
if !ok || volLen == 0 {
return ""
}
n, _ := v.next[id] // It's okay if this fails, n will be 0 and v.next[id] will be set to 1
if n >= volLen {
n = 0
}
v.next[id] = n + 1
return vol[n]
}
// Remove removes a volume from the volume cache.
func (v *VolumeCache) Remove(id uint32) {
v.Lock()
defer v.Unlock()
delete(v.volumeCache, id)
}
// Empty clears all data and returns the VolumeCache to it's initial state.
func (v *VolumeCache) Empty() {
v.Lock()
defer v.Unlock()
v.volumeCache = map[uint32][]string{}
v.next = map[uint32]int{}
}
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment