Add Versioning Server (#1576)
* Initial Webserver Draft for Version Controlling * Rename type to avoid confusion * Move Function Calls into Version Package * Fix Linting and Language Typos * Fix Linting and Spelling Mistakes * Include Copyright * Include Copyright * Adjust Version-Control Server to return list of Versions * Linting * Improve Request Handling and Readability * Add Configuration File Option Add Systemd Service file * Add Logging to File * Smaller Changes * Add Semantic Versioning and refuses outdated Software from Startup (#1612) * implements internal Semantic Version library * adds version logging + reporting to process * Advance SemVer struct for easier handling * Add Accepted Version Store * Fix Function * Restructure * Type Conversion * Handle Version String properly * Add Note about array index * Set temporary Default Version * Add Copyright * Adding Version to Dashboard * Adding Version Info Log * Renaming and adding CheckerProcess * Iteration Sync * Iteration V2 * linting * made LogAndReportVersion a go routine * Refactor to Go Routine * Add Context to Go Routine and allow Operation if Lookup to Control Server fails * Handle Unmarshal properly * Linting * Relocate Version Checks * Relocating Version Check and specified default Version for now * Linting Error Prevention * Refuse Startup on outdated Version * Add Startup Check Function * Straighten Logging * Dont force Shutdown if --dev flag is set * Create full Service/Peer Structure for ControlServer * Linting * Straighting Naming * Finish VersionControl Service Layout * Improve Error Handling * Change Listening Address * Move Checker Function * Remove VersionControl Peer * Linting * Linting * Create VersionClient Service * Renaming * Add Version Client to Peer Definitions * Linting and Renaming * Linting * Remove Transport Checks for now * Move to Client Side Flag * Remove check * Linting * Transport Client Version Intro * Adding Version Client to Transport Client * Add missing parameter * Adding Version Check, to set Allowed = true * Set Default to true, testing * Restructuring Code * Uplink Changes * Add more proper Defaults * Renaming of Version struct * Dont pass Service use Pointer * Set Defaults for Versioning Checks * Put HTTP Server in go routine * Add Versioncontrol to Storj-Sim * Testplanet Fixes * Linting * Add Error Handling and new Server Struct * Move Lock slightly * Reduce Race Potentials * Remove unnecessary files * Linting * Add Proper Transport Handling * small fixes * add fence for allowed check * Add Startup Version Check and Service Naming * make errormessage private * Add Comments about VersionedClient * Linting * Remove Checks that refuse outgoing connections * Remove release cmd * Add Release Script * Linting * Update to use correct Values * Move vars private and set minimum default versions for testing builds * Remove VersionedClient * Better Error Handling and naked return removal * Straighten the Regex and string conversion * Change Check to allows testplanet and storj-sim to run without the need to pass an LDFlag * Cosmetic Change to Dashboard * Cleanup Returns and remove commented code * Remove Version Check if no build options are passed in * Pass in Config Values instead of Pointers * Handle missed Error * Update Endpoint URL * Change Type of Release Flag * Add additional Logging * Remove Versions Logging of other Services * minor fixes Change-Id: I5cc04a410ea6b2008d14dffd63eb5f36dd348a8b
This commit is contained in:
Stefan Benten
GitHub
parent
f96724df91
commit
2cf86703a3
2
Makefile
2
Makefile
@ -76,7 +76,7 @@ proto: ## Rebuild protobuf files
|
||||
.PHONY: install-sim
|
||||
install-sim: ## install storj-sim
|
||||
@echo "Running ${@}"
|
||||
@go install -race -v storj.io/storj/cmd/storj-sim storj.io/storj/cmd/bootstrap storj.io/storj/cmd/satellite storj.io/storj/cmd/storagenode storj.io/storj/cmd/uplink storj.io/storj/cmd/gateway storj.io/storj/cmd/identity storj.io/storj/cmd/certificates
|
||||
@go install -race -v storj.io/storj/cmd/storj-sim storj.io/storj/cmd/versioncontrol storj.io/storj/cmd/bootstrap storj.io/storj/cmd/satellite storj.io/storj/cmd/storagenode storj.io/storj/cmd/uplink storj.io/storj/cmd/gateway storj.io/storj/cmd/identity storj.io/storj/cmd/certificates
|
||||
|
||||
##@ Test
|
||||
|
||||
|
||||
@ -15,6 +15,7 @@ import (
|
||||
|
||||
"storj.io/storj/bootstrap/bootstrapweb"
|
||||
"storj.io/storj/bootstrap/bootstrapweb/bootstrapserver"
|
||||
"storj.io/storj/internal/version"
|
||||
"storj.io/storj/pkg/identity"
|
||||
"storj.io/storj/pkg/kademlia"
|
||||
"storj.io/storj/pkg/pb"
|
||||
@ -44,6 +45,8 @@ type Config struct {
|
||||
Kademlia kademlia.Config
|
||||
|
||||
Web bootstrapserver.Config
|
||||
|
||||
Version version.Config
|
||||
}
|
||||
|
||||
// Verify verifies whether configuration is consistent and acceptable.
|
||||
@ -62,6 +65,8 @@ type Peer struct {
|
||||
|
||||
Server *server.Server
|
||||
|
||||
Version *version.Service
|
||||
|
||||
// services and endpoints
|
||||
Kademlia struct {
|
||||
RoutingTable *kademlia.RoutingTable
|
||||
@ -79,7 +84,7 @@ type Peer struct {
|
||||
}
|
||||
|
||||
// New creates a new Bootstrap Node.
|
||||
func New(log *zap.Logger, full *identity.FullIdentity, db DB, config Config) (*Peer, error) {
|
||||
func New(log *zap.Logger, full *identity.FullIdentity, db DB, config Config, versionInfo version.Info) (*Peer, error) {
|
||||
peer := &Peer{
|
||||
Log: log,
|
||||
Identity: full,
|
||||
@ -88,6 +93,15 @@ func New(log *zap.Logger, full *identity.FullIdentity, db DB, config Config) (*P
|
||||
|
||||
var err error
|
||||
|
||||
{
|
||||
test := version.Info{}
|
||||
if test != versionInfo {
|
||||
peer.Log.Sugar().Debugf("Binary Version: %s with CommitHash %s, built at %s as Release %v",
|
||||
versionInfo.Version.String(), versionInfo.CommitHash, versionInfo.Timestamp.String(), versionInfo.Release)
|
||||
peer.Version = version.NewService(config.Version, versionInfo, "Bootstrap")
|
||||
}
|
||||
}
|
||||
|
||||
{ // setup listener and server
|
||||
sc := config.Server
|
||||
options, err := tlsopts.NewOptions(peer.Identity, sc.Config)
|
||||
@ -174,6 +188,12 @@ func New(log *zap.Logger, full *identity.FullIdentity, db DB, config Config) (*P
|
||||
func (peer *Peer) Run(ctx context.Context) error {
|
||||
group, ctx := errgroup.WithContext(ctx)
|
||||
|
||||
group.Go(func() error {
|
||||
if peer.Version != nil {
|
||||
return ignoreCancel(peer.Version.Run(ctx))
|
||||
}
|
||||
return nil
|
||||
})
|
||||
group.Go(func() error {
|
||||
return ignoreCancel(peer.Kademlia.Service.Bootstrap(ctx))
|
||||
})
|
||||
|
||||
@ -15,6 +15,7 @@ import (
|
||||
"storj.io/storj/bootstrap"
|
||||
"storj.io/storj/bootstrap/bootstrapdb"
|
||||
"storj.io/storj/internal/fpath"
|
||||
"storj.io/storj/internal/version"
|
||||
"storj.io/storj/pkg/cfgstruct"
|
||||
"storj.io/storj/pkg/process"
|
||||
)
|
||||
@ -94,7 +95,7 @@ func cmdRun(cmd *cobra.Command, args []string) (err error) {
|
||||
return errs.New("Error creating tables for master database on bootstrap: %+v", err)
|
||||
}
|
||||
|
||||
peer, err := bootstrap.New(log, identity, db, runCfg)
|
||||
peer, err := bootstrap.New(log, identity, db, runCfg, version.Build)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@ -157,6 +157,7 @@ func cmdRun(cmd *cobra.Command, args []string) (err error) {
|
||||
if err := process.InitMetricsWithCertPath(ctx, nil, runCfg.Identity.CertPath); err != nil {
|
||||
zap.S().Error("Failed to initialize telemetry batcher: ", err)
|
||||
}
|
||||
|
||||
_, err = metainfo.ListBuckets(ctx, storj.BucketListOptions{Direction: storj.After})
|
||||
if err != nil {
|
||||
return fmt.Errorf("Failed to contact Satellite.\n"+
|
||||
|
||||
@ -16,6 +16,7 @@ import (
|
||||
"go.uber.org/zap"
|
||||
|
||||
"storj.io/storj/internal/fpath"
|
||||
"storj.io/storj/internal/version"
|
||||
"storj.io/storj/pkg/cfgstruct"
|
||||
"storj.io/storj/pkg/process"
|
||||
"storj.io/storj/satellite"
|
||||
@ -133,7 +134,7 @@ func cmdRun(cmd *cobra.Command, args []string) (err error) {
|
||||
return errs.New("Error creating tables for master database on satellite: %+v", err)
|
||||
}
|
||||
|
||||
peer, err := satellite.New(log, identity, db, &runCfg.Config)
|
||||
peer, err := satellite.New(log, identity, db, &runCfg.Config, version.Build)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@ -19,6 +19,7 @@ import (
|
||||
"go.uber.org/zap"
|
||||
|
||||
"storj.io/storj/internal/memory"
|
||||
"storj.io/storj/internal/version"
|
||||
"storj.io/storj/pkg/pb"
|
||||
"storj.io/storj/pkg/process"
|
||||
"storj.io/storj/pkg/transport"
|
||||
@ -80,7 +81,7 @@ func printDashboard(data *pb.DashboardResponse) error {
|
||||
color.NoColor = !useColor
|
||||
|
||||
heading := color.New(color.FgGreen, color.Bold)
|
||||
_, _ = heading.Printf("\nStorage Node Dashboard\n")
|
||||
_, _ = heading.Printf("\nStorage Node Dashboard ( Node Version: %s )\n", version.Build.Version.String())
|
||||
_, _ = heading.Printf("\n======================\n\n")
|
||||
|
||||
w := tabwriter.NewWriter(color.Output, 0, 0, 1, ' ', 0)
|
||||
|
||||
@ -15,6 +15,7 @@ import (
|
||||
"go.uber.org/zap"
|
||||
|
||||
"storj.io/storj/internal/fpath"
|
||||
"storj.io/storj/internal/version"
|
||||
"storj.io/storj/pkg/cfgstruct"
|
||||
"storj.io/storj/pkg/pb"
|
||||
"storj.io/storj/pkg/process"
|
||||
@ -147,7 +148,7 @@ func cmdRun(cmd *cobra.Command, args []string) (err error) {
|
||||
return errs.New("Error creating tables for master database on storagenode: %+v", err)
|
||||
}
|
||||
|
||||
peer, err := storagenode.New(log, identity, db, runCfg.Config)
|
||||
peer, err := storagenode.New(log, identity, db, runCfg.Config, version.Build)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@ -145,8 +145,27 @@ func newNetwork(flags *Flags) (*Processes, error) {
|
||||
storageNodePrivatePort = 13000
|
||||
consolePort = 10100
|
||||
bootstrapWebPort = 10010
|
||||
versioncontrolPort = 10011
|
||||
)
|
||||
|
||||
versioncontrol := processes.New(Info{
|
||||
Name: "versioncontrol/0",
|
||||
Executable: "versioncontrol",
|
||||
Directory: filepath.Join(processes.Directory, "versioncontrol", "0"),
|
||||
Address: net.JoinHostPort(host, strconv.Itoa(versioncontrolPort)),
|
||||
})
|
||||
|
||||
versioncontrol.Arguments = withCommon(versioncontrol.Directory, Arguments{
|
||||
"setup": {
|
||||
"--address", versioncontrol.Address,
|
||||
},
|
||||
"run": {},
|
||||
})
|
||||
|
||||
versioncontrol.ExecBefore["run"] = func(process *Process) error {
|
||||
return readConfigString(&versioncontrol.Address, versioncontrol.Directory, "address")
|
||||
}
|
||||
|
||||
bootstrap := processes.New(Info{
|
||||
Name: "bootstrap/0",
|
||||
Executable: "bootstrap",
|
||||
@ -154,6 +173,9 @@ func newNetwork(flags *Flags) (*Processes, error) {
|
||||
Address: net.JoinHostPort(host, strconv.Itoa(bootstrapPort)),
|
||||
})
|
||||
|
||||
// gateway must wait for the versioncontrol to start up
|
||||
bootstrap.WaitForStart(versioncontrol)
|
||||
|
||||
bootstrap.Arguments = withCommon(bootstrap.Directory, Arguments{
|
||||
"setup": {
|
||||
"--identity-dir", bootstrap.Directory,
|
||||
@ -169,6 +191,8 @@ func newNetwork(flags *Flags) (*Processes, error) {
|
||||
|
||||
"--server.extensions.revocation=false",
|
||||
"--server.use-peer-ca-whitelist=false",
|
||||
|
||||
"--version.server-address", fmt.Sprintf("http://%s/", versioncontrol.Address),
|
||||
},
|
||||
"run": {},
|
||||
})
|
||||
@ -217,6 +241,8 @@ func newNetwork(flags *Flags) (*Processes, error) {
|
||||
"--mail.smtp-server-address", "smtp.gmail.com:587",
|
||||
"--mail.from", "Storj <yaroslav-satellite-test@storj.io>",
|
||||
"--mail.template-path", filepath.Join(storjRoot, "web/satellite/static/emails"),
|
||||
|
||||
"--version.server-address", fmt.Sprintf("http://%s/", versioncontrol.Address),
|
||||
},
|
||||
"run": {},
|
||||
})
|
||||
@ -348,6 +374,8 @@ func newNetwork(flags *Flags) (*Processes, error) {
|
||||
"--server.extensions.revocation=false",
|
||||
"--server.use-peer-ca-whitelist=false",
|
||||
"--storage.satellite-id-restriction=false",
|
||||
|
||||
"--version.server-address", fmt.Sprintf("http://%s/", versioncontrol.Address),
|
||||
},
|
||||
"run": {},
|
||||
})
|
||||
|
||||
@ -44,7 +44,6 @@ func mountBucket(cmd *cobra.Command, args []string) (err error) {
|
||||
}
|
||||
|
||||
ctx := process.Ctx(cmd)
|
||||
|
||||
metainfo, streams, err := cfg.Metainfo(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
97
cmd/versioncontrol/main.go
Normal file
97
cmd/versioncontrol/main.go
Normal file
@ -0,0 +1,97 @@
|
||||
// Copyright (C) 2019 Storj Labs, Inc.
|
||||
// See LICENSE for copying information.
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
"go.uber.org/zap"
|
||||
|
||||
"storj.io/storj/internal/fpath"
|
||||
"storj.io/storj/pkg/cfgstruct"
|
||||
"storj.io/storj/pkg/process"
|
||||
"storj.io/storj/versioncontrol"
|
||||
)
|
||||
|
||||
var (
|
||||
rootCmd = &cobra.Command{
|
||||
Use: "versioncontrol",
|
||||
Short: "versioncontrol",
|
||||
}
|
||||
runCmd = &cobra.Command{
|
||||
Use: "run",
|
||||
Short: "Run the versioncontrol server",
|
||||
RunE: cmdRun,
|
||||
}
|
||||
setupCmd = &cobra.Command{
|
||||
Use: "setup",
|
||||
Short: "Create config files",
|
||||
RunE: cmdSetup,
|
||||
Annotations: map[string]string{"type": "setup"},
|
||||
}
|
||||
|
||||
runCfg versioncontrol.Config
|
||||
setupCfg versioncontrol.Config
|
||||
|
||||
confDir string
|
||||
isDev bool
|
||||
)
|
||||
|
||||
const (
|
||||
defaultServerAddr = ":8080"
|
||||
)
|
||||
|
||||
func init() {
|
||||
defaultConfDir := fpath.ApplicationDir("storj", "versioncontrol")
|
||||
cfgstruct.SetupFlag(zap.L(), rootCmd, &confDir, "config-dir", defaultConfDir, "main directory for versioncontrol configuration")
|
||||
cfgstruct.DevFlag(rootCmd, &isDev, false, "use development and test configuration settings")
|
||||
rootCmd.AddCommand(runCmd)
|
||||
rootCmd.AddCommand(setupCmd)
|
||||
cfgstruct.Bind(runCmd.Flags(), &runCfg, isDev, cfgstruct.ConfDir(confDir))
|
||||
cfgstruct.BindSetup(setupCmd.Flags(), &setupCfg, isDev, cfgstruct.ConfDir(confDir))
|
||||
}
|
||||
|
||||
func cmdRun(cmd *cobra.Command, args []string) (err error) {
|
||||
log := zap.L()
|
||||
controlserver, err := versioncontrol.New(log, &runCfg)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ctx := process.Ctx(cmd)
|
||||
err = controlserver.Run(ctx)
|
||||
return err
|
||||
}
|
||||
|
||||
func cmdSetup(cmd *cobra.Command, args []string) (err error) {
|
||||
setupDir, err := filepath.Abs(confDir)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
valid, _ := fpath.IsValidSetupDir(setupDir)
|
||||
if !valid {
|
||||
return fmt.Errorf("versioncontrol configuration already exists (%v)", setupDir)
|
||||
}
|
||||
|
||||
err = os.MkdirAll(setupDir, 0700)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
overrides := map[string]interface{}{}
|
||||
|
||||
serverAddress := cmd.Flag("address")
|
||||
if !serverAddress.Changed {
|
||||
overrides[serverAddress.Name] = defaultServerAddr
|
||||
}
|
||||
|
||||
return process.SaveConfigWithAllDefaults(cmd.Flags(), filepath.Join(setupDir, "config.yaml"), overrides)
|
||||
}
|
||||
|
||||
func main() {
|
||||
process.Exec(rootCmd)
|
||||
}
|
||||
14
cmd/versioncontrol/version-control.service
Normal file
14
cmd/versioncontrol/version-control.service
Normal file
@ -0,0 +1,14 @@
|
||||
[Unit]
|
||||
Description = Version Control service
|
||||
After = syslog.target
|
||||
|
||||
[Service]
|
||||
User = storj
|
||||
Group = storj
|
||||
ExecStart = /usr/local/bin/versioncontrol run -config-dir /etc/storj/versioncontrol/
|
||||
Restart = always
|
||||
Type = simple
|
||||
NotifyAccess = main
|
||||
|
||||
[Install]
|
||||
WantedBy = multi-user.target
|
||||
@ -7,6 +7,7 @@ package testplanet
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"net"
|
||||
@ -27,6 +28,7 @@ import (
|
||||
"storj.io/storj/bootstrap/bootstrapdb"
|
||||
"storj.io/storj/bootstrap/bootstrapweb/bootstrapserver"
|
||||
"storj.io/storj/internal/memory"
|
||||
"storj.io/storj/internal/version"
|
||||
"storj.io/storj/pkg/accounting/rollup"
|
||||
"storj.io/storj/pkg/accounting/tally"
|
||||
"storj.io/storj/pkg/audit"
|
||||
@ -53,6 +55,7 @@ import (
|
||||
"storj.io/storj/storagenode/orders"
|
||||
"storj.io/storj/storagenode/piecestore"
|
||||
"storj.io/storj/storagenode/storagenodedb"
|
||||
"storj.io/storj/versioncontrol"
|
||||
)
|
||||
|
||||
// Peer represents one of StorageNode or Satellite
|
||||
@ -89,6 +92,7 @@ type Planet struct {
|
||||
uplinks []*Uplink
|
||||
|
||||
Bootstrap *bootstrap.Peer
|
||||
VersionControl *versioncontrol.Peer
|
||||
Satellites []*satellite.Peer
|
||||
StorageNodes []*storagenode.Peer
|
||||
Uplinks []*Uplink
|
||||
@ -164,6 +168,11 @@ func NewCustom(log *zap.Logger, config Config) (*Planet, error) {
|
||||
}
|
||||
planet.whitelistPath = whitelistPath
|
||||
|
||||
planet.VersionControl, err = planet.newVersionControlServer()
|
||||
if err != nil {
|
||||
return nil, errs.Combine(err, planet.Shutdown())
|
||||
}
|
||||
|
||||
planet.Bootstrap, err = planet.newBootstrap()
|
||||
if err != nil {
|
||||
return nil, errs.Combine(err, planet.Shutdown())
|
||||
@ -210,6 +219,10 @@ func (planet *Planet) Start(ctx context.Context) {
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
planet.cancel = cancel
|
||||
|
||||
planet.run.Go(func() error {
|
||||
return planet.VersionControl.Run(ctx)
|
||||
})
|
||||
|
||||
for i := range planet.peers {
|
||||
peer := &planet.peers[i]
|
||||
peer.ctx, peer.cancel = context.WithCancel(ctx)
|
||||
@ -307,7 +320,6 @@ func (planet *Planet) Shutdown() error {
|
||||
case <-ctx.Done():
|
||||
}
|
||||
}()
|
||||
|
||||
errlist.Add(planet.run.Wait())
|
||||
cancel()
|
||||
|
||||
@ -323,6 +335,7 @@ func (planet *Planet) Shutdown() error {
|
||||
for _, db := range planet.databases {
|
||||
errlist.Add(db.Close())
|
||||
}
|
||||
errlist.Add(planet.VersionControl.Close())
|
||||
|
||||
errlist.Add(os.RemoveAll(planet.directory))
|
||||
return errlist.Err()
|
||||
@ -459,6 +472,7 @@ func (planet *Planet) newSatellites(count int) ([]*satellite.Peer, error) {
|
||||
Address: "127.0.0.1:0",
|
||||
PasswordCost: console.TestPasswordCost,
|
||||
},
|
||||
Version: planet.NewVersionConfig(),
|
||||
}
|
||||
if planet.config.Reconfigure.Satellite != nil {
|
||||
planet.config.Reconfigure.Satellite(log, i, &config)
|
||||
@ -475,7 +489,9 @@ func (planet *Planet) newSatellites(count int) ([]*satellite.Peer, error) {
|
||||
config.Console.StaticDir = filepath.Join(storjRoot, "web/satellite")
|
||||
config.Mail.TemplatePath = filepath.Join(storjRoot, "web/satellite/static/emails")
|
||||
|
||||
peer, err := satellite.New(log, identity, db, &config)
|
||||
verInfo := planet.NewVersionInfo()
|
||||
|
||||
peer, err := satellite.New(log, identity, db, &config, verInfo)
|
||||
if err != nil {
|
||||
return xs, err
|
||||
}
|
||||
@ -568,12 +584,15 @@ func (planet *Planet) newStorageNodes(count int, whitelistedSatelliteIDs []strin
|
||||
Timeout: time.Hour,
|
||||
},
|
||||
},
|
||||
Version: planet.NewVersionConfig(),
|
||||
}
|
||||
if planet.config.Reconfigure.StorageNode != nil {
|
||||
planet.config.Reconfigure.StorageNode(i, &config)
|
||||
}
|
||||
|
||||
peer, err := storagenode.New(log, identity, db, config)
|
||||
verInfo := planet.NewVersionInfo()
|
||||
|
||||
peer, err := storagenode.New(log, identity, db, config, verInfo)
|
||||
if err != nil {
|
||||
return xs, err
|
||||
}
|
||||
@ -645,12 +664,16 @@ func (planet *Planet) newBootstrap() (peer *bootstrap.Peer, err error) {
|
||||
Address: "127.0.0.1:0",
|
||||
StaticDir: "./web/bootstrap", // TODO: for development only
|
||||
},
|
||||
Version: planet.NewVersionConfig(),
|
||||
}
|
||||
if planet.config.Reconfigure.Bootstrap != nil {
|
||||
planet.config.Reconfigure.Bootstrap(0, &config)
|
||||
}
|
||||
|
||||
peer, err = bootstrap.New(log, identity, db, config)
|
||||
var verInfo version.Info
|
||||
verInfo = planet.NewVersionInfo()
|
||||
|
||||
peer, err = bootstrap.New(log, identity, db, config, verInfo)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@ -660,6 +683,60 @@ func (planet *Planet) newBootstrap() (peer *bootstrap.Peer, err error) {
|
||||
return peer, nil
|
||||
}
|
||||
|
||||
// newVersionControlServer initializes the Versioning Server
|
||||
func (planet *Planet) newVersionControlServer() (peer *versioncontrol.Peer, err error) {
|
||||
|
||||
prefix := "versioncontrol"
|
||||
log := planet.log.Named(prefix)
|
||||
dbDir := filepath.Join(planet.directory, prefix)
|
||||
|
||||
if err := os.MkdirAll(dbDir, 0700); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
config := &versioncontrol.Config{
|
||||
Address: "127.0.0.1:0",
|
||||
Versions: versioncontrol.ServiceVersions{
|
||||
Bootstrap: "v0.0.1",
|
||||
Satellite: "v0.0.1",
|
||||
Storagenode: "v0.0.1",
|
||||
Uplink: "v0.0.1",
|
||||
Gateway: "v0.0.1",
|
||||
},
|
||||
}
|
||||
peer, err = versioncontrol.New(log, config)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
log.Debug(" addr= " + peer.Addr())
|
||||
|
||||
return peer, nil
|
||||
}
|
||||
|
||||
// NewVersionInfo returns the Version Info for this planet with tuned metrics.
|
||||
func (planet *Planet) NewVersionInfo() version.Info {
|
||||
info := version.Info{
|
||||
Timestamp: time.Now(),
|
||||
CommitHash: "",
|
||||
Version: version.SemVer{
|
||||
Major: 0,
|
||||
Minor: 0,
|
||||
Patch: 1},
|
||||
Release: false,
|
||||
}
|
||||
return info
|
||||
}
|
||||
|
||||
// NewVersionConfig returns the Version Config for this planet with tuned metrics.
|
||||
func (planet *Planet) NewVersionConfig() version.Config {
|
||||
return version.Config{
|
||||
ServerAddress: fmt.Sprintf("http://%s/", planet.VersionControl.Addr()),
|
||||
RequestTimeout: time.Second * 15,
|
||||
CheckInterval: time.Minute * 5,
|
||||
}
|
||||
}
|
||||
|
||||
// Identities returns the identity provider for this planet.
|
||||
func (planet *Planet) Identities() *Identities {
|
||||
return planet.identities
|
||||
|
||||
168
internal/version/service.go
Normal file
168
internal/version/service.go
Normal file
@ -0,0 +1,168 @@
|
||||
// Copyright (C) 2019 Storj Labs, Inc.
|
||||
// See LICENSE for copying information.
|
||||
|
||||
package version
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"reflect"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/zeebo/errs"
|
||||
"go.uber.org/zap"
|
||||
|
||||
"storj.io/storj/internal/sync2"
|
||||
)
|
||||
|
||||
const (
|
||||
errOldVersion = "Outdated Software Version, please update!"
|
||||
)
|
||||
|
||||
// Config contains the necessary Information to check the Software Version
|
||||
type Config struct {
|
||||
ServerAddress string `help:"server address to check its version against" default:"https://version.alpha.storj.io"`
|
||||
RequestTimeout time.Duration `help:"Request timeout for version checks" default:"0h1m0s"`
|
||||
CheckInterval time.Duration `help:"Interval to check the version" default:"0h15m0s"`
|
||||
}
|
||||
|
||||
// Service contains the information and variables to ensure the Software is up to date
|
||||
type Service struct {
|
||||
config Config
|
||||
info Info
|
||||
service string
|
||||
|
||||
Loop *sync2.Cycle
|
||||
|
||||
checked chan struct{}
|
||||
mu sync.Mutex
|
||||
allowed bool
|
||||
}
|
||||
|
||||
// NewService creates a Version Check Client with default configuration
|
||||
func NewService(config Config, info Info, service string) (client *Service) {
|
||||
return &Service{
|
||||
config: config,
|
||||
info: info,
|
||||
service: service,
|
||||
Loop: sync2.NewCycle(config.CheckInterval),
|
||||
checked: make(chan struct{}, 0),
|
||||
allowed: false,
|
||||
}
|
||||
}
|
||||
|
||||
// Run logs the current version information
|
||||
func (srv *Service) Run(ctx context.Context) error {
|
||||
firstCheck := true
|
||||
return srv.Loop.Run(ctx, func(ctx context.Context) error {
|
||||
var err error
|
||||
allowed, err := srv.checkVersion(ctx)
|
||||
if err != nil {
|
||||
// Log about the error, but dont crash the service and allow further operation
|
||||
zap.S().Errorf("Failed to do periodic version check: ", err)
|
||||
allowed = true
|
||||
}
|
||||
|
||||
srv.mu.Lock()
|
||||
srv.allowed = allowed
|
||||
srv.mu.Unlock()
|
||||
|
||||
if firstCheck {
|
||||
close(srv.checked)
|
||||
firstCheck = false
|
||||
if !allowed {
|
||||
zap.S().Fatal(errOldVersion)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
// IsUpToDate returns whether if the Service is allowed to operate or not
|
||||
func (srv *Service) IsUpToDate() bool {
|
||||
<-srv.checked
|
||||
|
||||
srv.mu.Lock()
|
||||
defer srv.mu.Unlock()
|
||||
|
||||
return srv.allowed
|
||||
}
|
||||
|
||||
// CheckVersion checks if the client is running latest/allowed code
|
||||
func (srv *Service) checkVersion(ctx context.Context) (allowed bool, err error) {
|
||||
defer mon.Task()(&ctx)(&err)
|
||||
accepted, err := srv.queryVersionFromControlServer(ctx)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
list := getFieldString(&accepted, srv.service)
|
||||
zap.S().Debugf("allowed versions from Control Server: %v", list)
|
||||
|
||||
if list == nil {
|
||||
return true, errs.New("Empty List from Versioning Server")
|
||||
}
|
||||
if containsVersion(list, srv.info.Version) {
|
||||
zap.S().Infof("running on version %s", srv.info.Version.String())
|
||||
allowed = true
|
||||
} else {
|
||||
zap.S().Errorf("running on not allowed/outdated version %s", srv.info.Version.String())
|
||||
allowed = false
|
||||
}
|
||||
return allowed, err
|
||||
}
|
||||
|
||||
// QueryVersionFromControlServer handles the HTTP request to gather the allowed and latest version information
|
||||
func (srv *Service) queryVersionFromControlServer(ctx context.Context) (ver AllowedVersions, err error) {
|
||||
// Tune Client to have a custom Timeout (reduces hanging software)
|
||||
client := http.Client{
|
||||
Timeout: srv.config.RequestTimeout,
|
||||
}
|
||||
|
||||
// New Request that used the passed in context
|
||||
req, err := http.NewRequest("GET", srv.config.ServerAddress, nil)
|
||||
if err != nil {
|
||||
return AllowedVersions{}, err
|
||||
}
|
||||
req = req.WithContext(ctx)
|
||||
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return AllowedVersions{}, err
|
||||
}
|
||||
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
err = json.NewDecoder(resp.Body).Decode(&ver)
|
||||
return ver, err
|
||||
}
|
||||
|
||||
// DebugHandler returns a json representation of the current version information for the binary
|
||||
func (srv *Service) DebugHandler(w http.ResponseWriter, r *http.Request) {
|
||||
j, err := Build.Marshal()
|
||||
if err != nil {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
|
||||
_, err = w.Write(j)
|
||||
if err != nil {
|
||||
zap.S().Errorf("error writing data to client %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func getFieldString(array *AllowedVersions, field string) []SemVer {
|
||||
r := reflect.ValueOf(array)
|
||||
f := reflect.Indirect(r).FieldByName(field).Interface()
|
||||
result, ok := f.([]SemVer)
|
||||
if ok {
|
||||
return result
|
||||
}
|
||||
return nil
|
||||
}
|
||||
156
internal/version/version.go
Normal file
156
internal/version/version.go
Normal file
@ -0,0 +1,156 @@
|
||||
// Copyright (C) 2019 Storj Labs, Inc.
|
||||
// See LICENSE for copying information.
|
||||
|
||||
package version
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
monkit "gopkg.in/spacemonkeygo/monkit.v2"
|
||||
)
|
||||
|
||||
var (
|
||||
mon = monkit.Package()
|
||||
|
||||
// the following fields are set by linker flags. if any of them
|
||||
// are set and fail to parse, the program will fail to start
|
||||
buildTimestamp string // unix seconds since epoch
|
||||
buildCommitHash string
|
||||
buildVersion string // semantic version format
|
||||
buildRelease string // true/false
|
||||
|
||||
// Build is a struct containing all relevant build information associated with the binary
|
||||
Build Info
|
||||
)
|
||||
|
||||
// Info is the versioning information for a binary
|
||||
type Info struct {
|
||||
Timestamp time.Time `json:"timestamp,omitempty"`
|
||||
CommitHash string `json:"commitHash,omitempty"`
|
||||
Version SemVer `json:"version"`
|
||||
Release bool `json:"release,omitempty"`
|
||||
}
|
||||
|
||||
// SemVer represents a semantic version
|
||||
type SemVer struct {
|
||||
Major int64 `json:"major"`
|
||||
Minor int64 `json:"minor"`
|
||||
Patch int64 `json:"patch"`
|
||||
}
|
||||
|
||||
// AllowedVersions provides a list of SemVer per Service
|
||||
type AllowedVersions struct {
|
||||
Bootstrap []SemVer
|
||||
Satellite []SemVer
|
||||
Storagenode []SemVer
|
||||
Uplink []SemVer
|
||||
Gateway []SemVer
|
||||
}
|
||||
|
||||
// SemVerRegex is the regular expression used to parse a semantic version.
|
||||
// https://github.com/Masterminds/semver/blob/master/LICENSE.txt
|
||||
const SemVerRegex string = `v?([0-9]+)\.([0-9]+)\.([0-9]+)`
|
||||
|
||||
var versionRegex = regexp.MustCompile("^" + SemVerRegex + "$")
|
||||
|
||||
// NewSemVer parses a given version and returns an instance of SemVer or
|
||||
// an error if unable to parse the version.
|
||||
func NewSemVer(v string) (*SemVer, error) {
|
||||
m := versionRegex.FindStringSubmatch(v)
|
||||
if m == nil {
|
||||
return nil, errors.New("invalid semantic version for build")
|
||||
}
|
||||
|
||||
sv := SemVer{}
|
||||
|
||||
var err error
|
||||
|
||||
// first entry of m is the entire version string
|
||||
sv.Major, err = strconv.ParseInt(m[1], 10, 64)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
sv.Minor, err = strconv.ParseInt(m[2], 10, 64)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
sv.Patch, err = strconv.ParseInt(m[3], 10, 64)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &sv, nil
|
||||
}
|
||||
|
||||
// String converts the SemVer struct to a more easy to handle string
|
||||
func (sem *SemVer) String() (version string) {
|
||||
return fmt.Sprintf("v%d.%d.%d", sem.Major, sem.Minor, sem.Patch)
|
||||
}
|
||||
|
||||
// New creates Version_Info from a json byte array
|
||||
func New(data []byte) (v Info, err error) {
|
||||
err = json.Unmarshal(data, &v)
|
||||
return v, err
|
||||
}
|
||||
|
||||
// Marshal converts the existing Version Info to any json byte array
|
||||
func (v Info) Marshal() (data []byte, err error) {
|
||||
data, err = json.Marshal(v)
|
||||
return
|
||||
}
|
||||
|
||||
// containsVersion compares the allowed version array against the passed version
|
||||
func containsVersion(all []SemVer, x SemVer) bool {
|
||||
for _, n := range all {
|
||||
if x == n {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// StrToSemVerList converts a list of versions to a list of SemVer
|
||||
func StrToSemVerList(serviceVersions []string) (versions []SemVer, err error) {
|
||||
for _, subversion := range serviceVersions {
|
||||
sVer, err := NewSemVer(subversion)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
versions = append(versions, *sVer)
|
||||
}
|
||||
return versions, err
|
||||
}
|
||||
|
||||
func init() {
|
||||
if buildVersion == "" && buildTimestamp == "" && buildCommitHash == "" && buildRelease == "" {
|
||||
return
|
||||
}
|
||||
timestamp, err := strconv.ParseInt(buildTimestamp, 10, 64)
|
||||
if err != nil {
|
||||
panic(fmt.Sprintf("invalid timestamp: %v", err))
|
||||
}
|
||||
Build = Info{
|
||||
Timestamp: time.Unix(timestamp, 0),
|
||||
CommitHash: buildCommitHash,
|
||||
Release: strings.ToLower(buildRelease) == "true",
|
||||
}
|
||||
|
||||
sv, err := NewSemVer(buildVersion)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
Build.Version = *sv
|
||||
|
||||
if Build.Timestamp.Unix() == 0 || Build.CommitHash == "" {
|
||||
Build.Release = false
|
||||
}
|
||||
}
|
||||
@ -653,6 +653,7 @@ func TestCertificateSigner_Sign_E2E(t *testing.T) {
|
||||
|
||||
tlsOptions, err := tlsopts.NewOptions(clientIdent, tlsopts.Config{})
|
||||
require.NoError(t, err)
|
||||
|
||||
clientTransport := transport.NewClient(tlsOptions)
|
||||
|
||||
client, err := NewClient(ctx, clientTransport, service.Addr().String())
|
||||
@ -734,6 +735,7 @@ func TestNewClient(t *testing.T) {
|
||||
|
||||
tlsOptions, err := tlsopts.NewOptions(ident, tlsopts.Config{})
|
||||
require.NoError(t, err)
|
||||
|
||||
clientTransport := transport.NewClient(tlsOptions)
|
||||
|
||||
t.Run("Basic", func(t *testing.T) {
|
||||
|
||||
@ -527,6 +527,7 @@ func newKademlia(log *zap.Logger, nodeType pb.NodeType, bootstrapNodes []pb.Node
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
transportClient := transport.NewClient(tlsOptions, rt)
|
||||
|
||||
kadConfig := Config{
|
||||
|
||||
@ -32,6 +32,7 @@ func initDebug(logger *zap.Logger, r *monkit.Registry) (err error) {
|
||||
mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
|
||||
mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
|
||||
mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
|
||||
|
||||
mux.Handle("/mon/", http.StripPrefix("/mon", present.HTTP(r)))
|
||||
mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
|
||||
_, _ = fmt.Fprintln(w, "OK")
|
||||
|
||||
@ -9,6 +9,9 @@ import (
|
||||
"os"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
|
||||
// We use a blank import here to get the side effects from the init function in version
|
||||
_ "storj.io/storj/internal/version"
|
||||
)
|
||||
|
||||
func init() {
|
||||
|
||||
@ -51,8 +51,8 @@ func (client *slowTransport) Identity() *identity.FullIdentity {
|
||||
}
|
||||
|
||||
// WithObservers calls WithObservers for slowTransport
|
||||
func (client *slowTransport) WithObservers(obs ...Observer) *Transport {
|
||||
return client.client.WithObservers(obs...)
|
||||
func (client *slowTransport) WithObservers(obs ...Observer) Client {
|
||||
return &slowTransport{client.client.WithObservers(obs...), client.network}
|
||||
}
|
||||
|
||||
// DialOptions returns options such that it will use simulated network parameters
|
||||
|
||||
@ -26,7 +26,7 @@ type Client interface {
|
||||
DialNode(ctx context.Context, node *pb.Node, opts ...grpc.DialOption) (*grpc.ClientConn, error)
|
||||
DialAddress(ctx context.Context, address string, opts ...grpc.DialOption) (*grpc.ClientConn, error)
|
||||
Identity() *identity.FullIdentity
|
||||
WithObservers(obs ...Observer) *Transport
|
||||
WithObservers(obs ...Observer) Client
|
||||
}
|
||||
|
||||
// Transport interface structure
|
||||
@ -57,13 +57,13 @@ func NewClientWithTimeout(tlsOpts *tlsopts.Options, requestTimeout time.Duration
|
||||
// target node has the private key for the requested node ID.
|
||||
func (transport *Transport) DialNode(ctx context.Context, node *pb.Node, opts ...grpc.DialOption) (conn *grpc.ClientConn, err error) {
|
||||
defer mon.Task()(&ctx)(&err)
|
||||
|
||||
if node != nil {
|
||||
node.Type.DPanicOnInvalid("transport dial node")
|
||||
}
|
||||
if node.Address == nil || node.Address.Address == "" {
|
||||
return nil, Error.New("no address")
|
||||
}
|
||||
|
||||
dialOption, err := transport.tlsOpts.DialOption(node.Id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@ -124,7 +124,7 @@ func (transport *Transport) Identity() *identity.FullIdentity {
|
||||
}
|
||||
|
||||
// WithObservers returns a new transport including the listed observers.
|
||||
func (transport *Transport) WithObservers(obs ...Observer) *Transport {
|
||||
func (transport *Transport) WithObservers(obs ...Observer) Client {
|
||||
tr := &Transport{tlsOpts: transport.tlsOpts, requestTimeout: transport.requestTimeout}
|
||||
tr.observers = append(tr.observers, transport.observers...)
|
||||
tr.observers = append(tr.observers, obs...)
|
||||
|
||||
@ -21,6 +21,7 @@ import (
|
||||
|
||||
"storj.io/storj/internal/post"
|
||||
"storj.io/storj/internal/post/oauth2"
|
||||
"storj.io/storj/internal/version"
|
||||
"storj.io/storj/pkg/accounting"
|
||||
"storj.io/storj/pkg/accounting/rollup"
|
||||
"storj.io/storj/pkg/accounting/tally"
|
||||
@ -109,6 +110,8 @@ type Config struct {
|
||||
|
||||
Mail mailservice.Config
|
||||
Console consoleweb.Config
|
||||
|
||||
Version version.Config
|
||||
}
|
||||
|
||||
// Peer is the satellite
|
||||
@ -122,6 +125,8 @@ type Peer struct {
|
||||
|
||||
Server *server.Server
|
||||
|
||||
Version *version.Service
|
||||
|
||||
// services and endpoints
|
||||
Kademlia struct {
|
||||
kdb, ndb storage.KeyValueStore // TODO: move these into DB
|
||||
@ -186,7 +191,7 @@ type Peer struct {
|
||||
}
|
||||
|
||||
// New creates a new satellite
|
||||
func New(log *zap.Logger, full *identity.FullIdentity, db DB, config *Config) (*Peer, error) {
|
||||
func New(log *zap.Logger, full *identity.FullIdentity, db DB, config *Config, versionInfo version.Info) (*Peer, error) {
|
||||
peer := &Peer{
|
||||
Log: log,
|
||||
Identity: full,
|
||||
@ -195,6 +200,15 @@ func New(log *zap.Logger, full *identity.FullIdentity, db DB, config *Config) (*
|
||||
|
||||
var err error
|
||||
|
||||
{
|
||||
test := version.Info{}
|
||||
if test != versionInfo {
|
||||
peer.Log.Sugar().Debugf("Binary Version: %s with CommitHash %s, built at %s as Release %v",
|
||||
versionInfo.Version.String(), versionInfo.CommitHash, versionInfo.Timestamp.String(), versionInfo.Release)
|
||||
peer.Version = version.NewService(config.Version, versionInfo, "Satellite")
|
||||
}
|
||||
}
|
||||
|
||||
{ // setup listener and server
|
||||
log.Debug("Starting listener and server")
|
||||
sc := config.Server
|
||||
@ -513,6 +527,12 @@ func ignoreCancel(err error) error {
|
||||
func (peer *Peer) Run(ctx context.Context) error {
|
||||
group, ctx := errgroup.WithContext(ctx)
|
||||
|
||||
group.Go(func() error {
|
||||
if peer.Version != nil {
|
||||
return ignoreCancel(peer.Version.Run(ctx))
|
||||
}
|
||||
return nil
|
||||
})
|
||||
group.Go(func() error {
|
||||
return ignoreCancel(peer.Kademlia.Service.Bootstrap(ctx))
|
||||
})
|
||||
|
||||
29
scripts/release.sh
Executable file
29
scripts/release.sh
Executable file
@ -0,0 +1,29 @@
|
||||
#!/usr/bin/env bash
|
||||
|
||||
set -eu
|
||||
set -o pipefail
|
||||
|
||||
echo -n "Build timestamp: "
|
||||
TIMESTAMP=$(date +%s)
|
||||
echo $TIMESTAMP
|
||||
|
||||
echo -n "Git commit: "
|
||||
if [[ "$(git diff --stat)" != '' ]] || [[ -n "$(git status -s)" ]]; then
|
||||
COMMIT=$(git rev-parse HEAD)-dirty
|
||||
RELEASE=false
|
||||
else
|
||||
COMMIT=$(git rev-parse HEAD)
|
||||
RELEASE=true
|
||||
fi
|
||||
echo $COMMIT
|
||||
|
||||
echo -n "Tagged version: "
|
||||
VERSION=$(git describe --tags --exact-match --match "v[0-9]*.[0-9]*.[0-9]*")
|
||||
echo $VERSION
|
||||
|
||||
echo Running "go $@"
|
||||
exec go "$1" -ldflags \
|
||||
"-X storj.io/storj/internal/version.buildTimestamp=$TIMESTAMP
|
||||
-X storj.io/storj/internal/version.buildCommitHash=$COMMIT
|
||||
-X storj.io/storj/internal/version.buildVersion=$VERSION
|
||||
-X storj.io/storj/internal/version.buildRelease=$RELEASE" "${@:2}"
|
||||
@ -11,6 +11,7 @@ import (
|
||||
"golang.org/x/sync/errgroup"
|
||||
"google.golang.org/grpc"
|
||||
|
||||
"storj.io/storj/internal/version"
|
||||
"storj.io/storj/pkg/auth/signing"
|
||||
"storj.io/storj/pkg/identity"
|
||||
"storj.io/storj/pkg/kademlia"
|
||||
@ -61,6 +62,8 @@ type Config struct {
|
||||
Storage psserver.Config
|
||||
|
||||
Storage2 piecestore.Config
|
||||
|
||||
Version version.Config
|
||||
}
|
||||
|
||||
// Verify verifies whether configuration is consistent and acceptable.
|
||||
@ -79,6 +82,8 @@ type Peer struct {
|
||||
|
||||
Server *server.Server
|
||||
|
||||
Version *version.Service
|
||||
|
||||
// services and endpoints
|
||||
// TODO: similar grouping to satellite.Peer
|
||||
Kademlia struct {
|
||||
@ -103,7 +108,7 @@ type Peer struct {
|
||||
}
|
||||
|
||||
// New creates a new Storage Node.
|
||||
func New(log *zap.Logger, full *identity.FullIdentity, db DB, config Config) (*Peer, error) {
|
||||
func New(log *zap.Logger, full *identity.FullIdentity, db DB, config Config, versionInfo version.Info) (*Peer, error) {
|
||||
peer := &Peer{
|
||||
Log: log,
|
||||
Identity: full,
|
||||
@ -112,6 +117,15 @@ func New(log *zap.Logger, full *identity.FullIdentity, db DB, config Config) (*P
|
||||
|
||||
var err error
|
||||
|
||||
{
|
||||
test := version.Info{}
|
||||
if test != versionInfo {
|
||||
peer.Log.Sugar().Debugf("Binary Version: %s with CommitHash %s, built at %s as Release %v",
|
||||
versionInfo.Version.String(), versionInfo.CommitHash, versionInfo.Timestamp.String(), versionInfo.Release)
|
||||
peer.Version = version.NewService(config.Version, versionInfo, "Storagenode")
|
||||
}
|
||||
}
|
||||
|
||||
{ // setup listener and server
|
||||
sc := config.Server
|
||||
options, err := tlsopts.NewOptions(peer.Identity, sc.Config)
|
||||
@ -238,6 +252,12 @@ func New(log *zap.Logger, full *identity.FullIdentity, db DB, config Config) (*P
|
||||
func (peer *Peer) Run(ctx context.Context) error {
|
||||
group, ctx := errgroup.WithContext(ctx)
|
||||
|
||||
group.Go(func() error {
|
||||
if peer.Version != nil {
|
||||
return ignoreCancel(peer.Version.Run(ctx))
|
||||
}
|
||||
return nil
|
||||
})
|
||||
group.Go(func() error {
|
||||
return ignoreCancel(peer.Kademlia.Service.Bootstrap(ctx))
|
||||
})
|
||||
|
||||
@ -78,6 +78,9 @@ func (c Config) GetMetainfo(ctx context.Context, identity *identity.FullIdentity
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
// ToDo: Handle Versioning for Uplinks here
|
||||
|
||||
tc := transport.NewClient(tlsOpts)
|
||||
|
||||
if c.Client.SatelliteAddr == "" {
|
||||
|
||||
146
versioncontrol/peer.go
Normal file
146
versioncontrol/peer.go
Normal file
@ -0,0 +1,146 @@
|
||||
// Copyright (C) 2019 Storj Labs, Inc.
|
||||
// See LICENSE for copying information.
|
||||
|
||||
package versioncontrol
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"github.com/zeebo/errs"
|
||||
"go.uber.org/zap"
|
||||
"golang.org/x/sync/errgroup"
|
||||
|
||||
"storj.io/storj/internal/version"
|
||||
)
|
||||
|
||||
// Config is all the configuration parameters for a Version Control Server
|
||||
type Config struct {
|
||||
Address string `user:"true" help:"public address to listen on" default:":8080"`
|
||||
Versions ServiceVersions
|
||||
}
|
||||
|
||||
// ServiceVersions provides a list of allowed Versions per Service
|
||||
type ServiceVersions struct {
|
||||
Bootstrap string `user:"true" help:"Allowed Bootstrap Versions" default:"v0.0.1"`
|
||||
Satellite string `user:"true" help:"Allowed Satellite Versions" default:"v0.0.1"`
|
||||
Storagenode string `user:"true" help:"Allowed Storagenode Versions" default:"v0.0.1"`
|
||||
Uplink string `user:"true" help:"Allowed Uplink Versions" default:"v0.0.1"`
|
||||
Gateway string `user:"true" help:"Allowed Gateway Versions" default:"v0.0.1"`
|
||||
}
|
||||
|
||||
// Peer is the representation of a VersionControl Server.
|
||||
type Peer struct {
|
||||
// core dependencies
|
||||
Log *zap.Logger
|
||||
|
||||
// Web server
|
||||
Server struct {
|
||||
Endpoint http.Server
|
||||
Listener net.Listener
|
||||
}
|
||||
Versions version.AllowedVersions
|
||||
|
||||
// response contains the byte version of current allowed versions
|
||||
response []byte
|
||||
}
|
||||
|
||||
func ignoreCancel(err error) error {
|
||||
if err == context.Canceled || err == http.ErrServerClosed {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// HandleGet contains the request handler for the version control web server
|
||||
func (peer *Peer) HandleGet(w http.ResponseWriter, r *http.Request) {
|
||||
// Only handle GET Requests
|
||||
if r.Method != http.MethodGet {
|
||||
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
|
||||
var xfor string
|
||||
if xfor = r.Header.Get("X-Forwarded-For"); xfor == "" {
|
||||
xfor = r.RemoteAddr
|
||||
}
|
||||
zap.S().Debugf("Request from: %s for %s", r.RemoteAddr, xfor)
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, err := w.Write(peer.response)
|
||||
if err != nil {
|
||||
zap.S().Errorf("error writing response to client: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// New creates a new VersionControl Server.
|
||||
func New(log *zap.Logger, config *Config) (peer *Peer, err error) {
|
||||
peer = &Peer{
|
||||
Log: log,
|
||||
}
|
||||
|
||||
// Convert each Service's Version String to List of SemVer
|
||||
bootstrapVersions := strings.Split(config.Versions.Bootstrap, ",")
|
||||
peer.Versions.Bootstrap, err = version.StrToSemVerList(bootstrapVersions)
|
||||
|
||||
satelliteVersions := strings.Split(config.Versions.Satellite, ",")
|
||||
peer.Versions.Satellite, err = version.StrToSemVerList(satelliteVersions)
|
||||
|
||||
storagenodeVersions := strings.Split(config.Versions.Storagenode, ",")
|
||||
peer.Versions.Storagenode, err = version.StrToSemVerList(storagenodeVersions)
|
||||
|
||||
uplinkVersions := strings.Split(config.Versions.Uplink, ",")
|
||||
peer.Versions.Uplink, err = version.StrToSemVerList(uplinkVersions)
|
||||
|
||||
gatewayVersions := strings.Split(config.Versions.Gateway, ",")
|
||||
peer.Versions.Gateway, err = version.StrToSemVerList(gatewayVersions)
|
||||
|
||||
peer.response, err = json.Marshal(peer.Versions)
|
||||
|
||||
if err != nil {
|
||||
peer.Log.Sugar().Fatalf("Error marshalling version info: %v", err)
|
||||
}
|
||||
|
||||
peer.Log.Sugar().Debugf("setting version info to: %v", string(peer.response))
|
||||
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/", peer.HandleGet)
|
||||
peer.Server.Endpoint = http.Server{
|
||||
Handler: mux,
|
||||
}
|
||||
|
||||
peer.Server.Listener, err = net.Listen("tcp", config.Address)
|
||||
if err != nil {
|
||||
return nil, errs.Combine(err, peer.Close())
|
||||
}
|
||||
return peer, nil
|
||||
}
|
||||
|
||||
// Run runs versioncontrol server until it's either closed or it errors.
|
||||
func (peer *Peer) Run(ctx context.Context) (err error) {
|
||||
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
var group errgroup.Group
|
||||
|
||||
group.Go(func() error {
|
||||
<-ctx.Done()
|
||||
return ignoreCancel(peer.Server.Endpoint.Shutdown(ctx))
|
||||
})
|
||||
group.Go(func() error {
|
||||
defer cancel()
|
||||
peer.Log.Sugar().Infof("Versioning server started on %s", peer.Addr())
|
||||
return ignoreCancel(peer.Server.Endpoint.Serve(peer.Server.Listener))
|
||||
})
|
||||
return group.Wait()
|
||||
}
|
||||
|
||||
// Close closes all the resources.
|
||||
func (peer *Peer) Close() (err error) {
|
||||
return peer.Server.Endpoint.Close()
|
||||
}
|
||||
|
||||
// Addr returns the public address.
|
||||
func (peer *Peer) Addr() string { return peer.Server.Listener.Addr().String() }
|
||||
Loading…
Reference in New Issue
Block a user