5075504926
The JSON parsing of the config could error, but we skipped error checking and use Validate() at the end. This caused that maybe some JSON parsing errors where logged but the final error when validating the configuration came from somewhere different, creating very confusing error messages for the user. This changes this, along with removing hardcoded section lists. This also removes a Sharder component section because, AFAIK, it was a left over from the sharding branch and in the end there is no separate sharding component that needs a config. License: MIT Signed-off-by: Hector Sanjuan <code@hector.link>
466 lines
11 KiB
Go
466 lines
11 KiB
Go
// Package config provides interfaces and utilities for different Cluster
|
|
// components to register, read, write and validate configuration sections
|
|
// stored in a central configuration file.
|
|
package config
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io/ioutil"
|
|
"path/filepath"
|
|
"sync"
|
|
"time"
|
|
|
|
logging "github.com/ipfs/go-log"
|
|
)
|
|
|
|
var logger = logging.Logger("config")
|
|
|
|
// ConfigSaveInterval specifies how often to save the configuration file if
|
|
// it needs saving.
|
|
var ConfigSaveInterval = time.Second
|
|
|
|
// The ComponentConfig interface allows components to define configurations
|
|
// which can be managed as part of the ipfs-cluster configuration file by the
|
|
// Manager.
|
|
type ComponentConfig interface {
|
|
// Returns a string identifying the section name for this configuration
|
|
ConfigKey() string
|
|
// Parses a JSON representation of this configuration
|
|
LoadJSON([]byte) error
|
|
// Provides a JSON representation of this configuration
|
|
ToJSON() ([]byte, error)
|
|
// Sets default working values
|
|
Default() error
|
|
// Allows this component to work under a subfolder
|
|
SetBaseDir(string)
|
|
// Checks that the configuration is valid
|
|
Validate() error
|
|
// Provides a channel to signal the Manager that the configuration
|
|
// should be persisted.
|
|
SaveCh() <-chan struct{}
|
|
}
|
|
|
|
// These are the component configuration types
|
|
// supported by the Manager.
|
|
const (
|
|
Cluster SectionType = iota
|
|
Consensus
|
|
API
|
|
IPFSConn
|
|
State
|
|
PinTracker
|
|
Monitor
|
|
Allocator
|
|
Informer
|
|
endTypes // keep this at the end
|
|
)
|
|
|
|
// SectionType specifies to which section a component configuration belongs.
|
|
type SectionType int
|
|
|
|
// SectionTypes returns the list of supported SectionTypes
|
|
func SectionTypes() []SectionType {
|
|
var l []SectionType
|
|
for i := Cluster; i < endTypes; i++ {
|
|
l = append(l, i)
|
|
}
|
|
return l
|
|
}
|
|
|
|
// Section is a section of which stores
|
|
// component-specific configurations.
|
|
type Section map[string]ComponentConfig
|
|
|
|
// jsonSection stores component specific
|
|
// configurations. Component configurations depend on
|
|
// components themselves.
|
|
type jsonSection map[string]*json.RawMessage
|
|
|
|
// Manager represents an ipfs-cluster configuration which bundles
|
|
// different ComponentConfigs object together.
|
|
// Use RegisterComponent() to add a component configurations to the
|
|
// object. Once registered, configurations will be parsed from the
|
|
// central configuration file when doing LoadJSON(), and saved to it
|
|
// when doing SaveJSON().
|
|
type Manager struct {
|
|
ctx context.Context
|
|
cancel func()
|
|
wg sync.WaitGroup
|
|
|
|
// The Cluster configuration has a top-level
|
|
// special section.
|
|
clusterConfig ComponentConfig
|
|
|
|
// Holds configuration objects for components.
|
|
sections map[SectionType]Section
|
|
|
|
// store originally parsed jsonConfig
|
|
jsonCfg *jsonConfig
|
|
|
|
// if a config has been loaded from disk, track the path
|
|
// so it can be saved to the same place.
|
|
path string
|
|
saveMux sync.Mutex
|
|
}
|
|
|
|
// NewManager returns a correctly initialized Manager
|
|
// which is ready to accept component configurations.
|
|
func NewManager() *Manager {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
return &Manager{
|
|
ctx: ctx,
|
|
cancel: cancel,
|
|
sections: make(map[SectionType]Section),
|
|
}
|
|
|
|
}
|
|
|
|
// Shutdown makes sure all configuration save operations are finished
|
|
// before returning.
|
|
func (cfg *Manager) Shutdown() {
|
|
cfg.cancel()
|
|
cfg.wg.Wait()
|
|
}
|
|
|
|
// this watches a save channel which is used to signal that
|
|
// we need to store changes in the configuration.
|
|
// because saving can be called too much, we will only
|
|
// save at intervals of 1 save/second at most.
|
|
func (cfg *Manager) watchSave(save <-chan struct{}) {
|
|
defer cfg.wg.Done()
|
|
|
|
// Save once per second mostly
|
|
ticker := time.NewTicker(ConfigSaveInterval)
|
|
defer ticker.Stop()
|
|
|
|
thingsToSave := false
|
|
|
|
for {
|
|
select {
|
|
case <-save:
|
|
thingsToSave = true
|
|
case <-ticker.C:
|
|
if thingsToSave {
|
|
err := cfg.SaveJSON("")
|
|
if err != nil {
|
|
logger.Error(err)
|
|
}
|
|
thingsToSave = false
|
|
}
|
|
|
|
// Exit if we have to
|
|
select {
|
|
case <-cfg.ctx.Done():
|
|
return
|
|
default:
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// jsonConfig represents a Cluster configuration as it will look when it is
|
|
// saved using json. Most configuration keys are converted into simple types
|
|
// like strings, and key names aim to be self-explanatory for the user.
|
|
type jsonConfig struct {
|
|
Cluster *json.RawMessage `json:"cluster"`
|
|
Consensus jsonSection `json:"consensus,omitempty"`
|
|
API jsonSection `json:"api,omitempty"`
|
|
IPFSConn jsonSection `json:"ipfs_connector,omitempty"`
|
|
State jsonSection `json:"state,omitempty"`
|
|
PinTracker jsonSection `json:"pin_tracker,omitempty"`
|
|
Monitor jsonSection `json:"monitor,omitempty"`
|
|
Allocator jsonSection `json:"allocator,omitempty"`
|
|
Informer jsonSection `json:"informer,omitempty"`
|
|
}
|
|
|
|
func (jcfg *jsonConfig) getSection(i SectionType) jsonSection {
|
|
switch i {
|
|
case Consensus:
|
|
return jcfg.Consensus
|
|
case API:
|
|
return jcfg.API
|
|
case IPFSConn:
|
|
return jcfg.IPFSConn
|
|
case State:
|
|
return jcfg.State
|
|
case PinTracker:
|
|
return jcfg.PinTracker
|
|
case Monitor:
|
|
return jcfg.Monitor
|
|
case Allocator:
|
|
return jcfg.Allocator
|
|
case Informer:
|
|
return jcfg.Informer
|
|
default:
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// Default generates a default configuration by generating defaults for all
|
|
// registered components.
|
|
func (cfg *Manager) Default() error {
|
|
for _, section := range cfg.sections {
|
|
for k, compcfg := range section {
|
|
logger.Debugf("generating default conf for %s", k)
|
|
err := compcfg.Default()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
if cfg.clusterConfig != nil {
|
|
logger.Debug("generating default conf for cluster")
|
|
err := cfg.clusterConfig.Default()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// RegisterComponent lets the Manager load and save component configurations
|
|
func (cfg *Manager) RegisterComponent(t SectionType, ccfg ComponentConfig) {
|
|
cfg.wg.Add(1)
|
|
go cfg.watchSave(ccfg.SaveCh())
|
|
|
|
if t == Cluster {
|
|
cfg.clusterConfig = ccfg
|
|
return
|
|
}
|
|
|
|
if cfg.sections == nil {
|
|
cfg.sections = make(map[SectionType]Section)
|
|
}
|
|
|
|
_, ok := cfg.sections[t]
|
|
if !ok {
|
|
cfg.sections[t] = make(Section)
|
|
}
|
|
|
|
cfg.sections[t][ccfg.ConfigKey()] = ccfg
|
|
}
|
|
|
|
// Validate checks that all the registered components in this
|
|
// Manager have valid configurations. It also makes sure that
|
|
// the main Cluster compoenent exists.
|
|
func (cfg *Manager) Validate() error {
|
|
if cfg.clusterConfig == nil {
|
|
return errors.New("no registered cluster section")
|
|
}
|
|
|
|
if cfg.sections == nil {
|
|
return errors.New("no registered components")
|
|
}
|
|
|
|
err := cfg.clusterConfig.Validate()
|
|
if err != nil {
|
|
return fmt.Errorf("cluster section failed to validate: %s", err)
|
|
}
|
|
|
|
for t, section := range cfg.sections {
|
|
if section == nil {
|
|
return fmt.Errorf("section %d is nil", t)
|
|
}
|
|
for k, compCfg := range section {
|
|
if compCfg == nil {
|
|
return fmt.Errorf("%s entry for section %d is nil", k, t)
|
|
}
|
|
err := compCfg.Validate()
|
|
if err != nil {
|
|
return fmt.Errorf("%s failed to validate: %s", k, err)
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// LoadJSONFromFile reads a Configuration file from disk and parses
|
|
// it. See LoadJSON too.
|
|
func (cfg *Manager) LoadJSONFromFile(path string) error {
|
|
cfg.path = path
|
|
|
|
file, err := ioutil.ReadFile(path)
|
|
if err != nil {
|
|
logger.Error("error reading the configuration file: ", err)
|
|
return err
|
|
}
|
|
|
|
err = cfg.LoadJSON(file)
|
|
return err
|
|
}
|
|
|
|
// LoadJSON parses configurations for all registered components,
|
|
// In order to work, component configurations must have been registered
|
|
// beforehand with RegisterComponent.
|
|
func (cfg *Manager) LoadJSON(bs []byte) error {
|
|
dir := filepath.Dir(cfg.path)
|
|
|
|
jcfg := &jsonConfig{}
|
|
err := json.Unmarshal(bs, jcfg)
|
|
if err != nil {
|
|
logger.Error("error parsing JSON: ", err)
|
|
return err
|
|
}
|
|
|
|
cfg.jsonCfg = jcfg
|
|
|
|
// Load Cluster section. Needs to have been registered
|
|
if cfg.clusterConfig != nil && jcfg.Cluster != nil {
|
|
cfg.clusterConfig.SetBaseDir(dir)
|
|
err = cfg.clusterConfig.LoadJSON([]byte(*jcfg.Cluster))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
loadCompJSON := func(name string, component ComponentConfig, jsonSection jsonSection) error {
|
|
raw, ok := jsonSection[name]
|
|
if ok {
|
|
component.SetBaseDir(dir)
|
|
err := component.LoadJSON([]byte(*raw))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
logger.Debugf("%s section configuration loaded", name)
|
|
} else {
|
|
logger.Warningf("%s section is empty, generating default", name)
|
|
component.SetBaseDir(dir)
|
|
component.Default()
|
|
}
|
|
|
|
return nil
|
|
}
|
|
// Helper function to load json from each section in the json config
|
|
loadSectionJSON := func(section Section, jsonSection jsonSection) error {
|
|
for name, component := range section {
|
|
err := loadCompJSON(name, component, jsonSection)
|
|
if err != nil {
|
|
logger.Error(err)
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
|
|
}
|
|
|
|
sections := cfg.sections
|
|
|
|
for _, t := range SectionTypes() {
|
|
if t == Cluster {
|
|
continue
|
|
}
|
|
err := loadSectionJSON(sections[t], jcfg.getSection(t))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
// Should we change hardcoded "ipfsproxy" to something else
|
|
if _, ok := jcfg.API["ipfsproxy"]; !ok {
|
|
loadCompJSON("ipfshttp", sections[API]["ipfsproxy"], jcfg.IPFSConn)
|
|
logger.Warning(`
|
|
The IPFS proxy functionality has been extracted as a separate component
|
|
and now uses its own configuration section ("ipfsproxy" in the "api" section).
|
|
|
|
To keep compatibility, since you did not define an "ipfsproxy" section, the
|
|
proxy configuration is taken from the "ipfshttp" section as before, but this
|
|
will be removed in future versions.
|
|
`)
|
|
}
|
|
return cfg.Validate()
|
|
}
|
|
|
|
// SaveJSON saves the JSON representation of the Config to
|
|
// the given path.
|
|
func (cfg *Manager) SaveJSON(path string) error {
|
|
cfg.saveMux.Lock()
|
|
defer cfg.saveMux.Unlock()
|
|
|
|
logger.Info("Saving configuration")
|
|
|
|
if path == "" {
|
|
path = cfg.path
|
|
}
|
|
|
|
bs, err := cfg.ToJSON()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return ioutil.WriteFile(path, bs, 0600)
|
|
}
|
|
|
|
// ToJSON provides a JSON representation of the configuration by
|
|
// generating JSON for all componenents registered.
|
|
func (cfg *Manager) ToJSON() ([]byte, error) {
|
|
err := cfg.Validate()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
jcfg := cfg.jsonCfg
|
|
if jcfg == nil {
|
|
jcfg = &jsonConfig{}
|
|
}
|
|
|
|
if cfg.clusterConfig != nil {
|
|
raw, err := cfg.clusterConfig.ToJSON()
|
|
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
jcfg.Cluster = new(json.RawMessage)
|
|
*jcfg.Cluster = raw
|
|
logger.Debug("writing changes for cluster section")
|
|
}
|
|
|
|
// Given a Section and a *jsonSection, it updates the
|
|
// component-configurations in the latter.
|
|
updateJSONConfigs := func(section Section, dest *jsonSection) error {
|
|
for k, v := range section {
|
|
logger.Debugf("writing changes for %s section", k)
|
|
j, err := v.ToJSON()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if *dest == nil {
|
|
*dest = make(jsonSection)
|
|
}
|
|
jsonSection := *dest
|
|
jsonSection[k] = new(json.RawMessage)
|
|
*jsonSection[k] = j
|
|
}
|
|
return nil
|
|
}
|
|
|
|
for k, v := range cfg.sections {
|
|
var err error
|
|
switch k {
|
|
case Consensus:
|
|
err = updateJSONConfigs(v, &jcfg.Consensus)
|
|
case API:
|
|
err = updateJSONConfigs(v, &jcfg.API)
|
|
case IPFSConn:
|
|
err = updateJSONConfigs(v, &jcfg.IPFSConn)
|
|
case State:
|
|
err = updateJSONConfigs(v, &jcfg.State)
|
|
case PinTracker:
|
|
err = updateJSONConfigs(v, &jcfg.PinTracker)
|
|
case Monitor:
|
|
err = updateJSONConfigs(v, &jcfg.Monitor)
|
|
case Allocator:
|
|
err = updateJSONConfigs(v, &jcfg.Allocator)
|
|
case Informer:
|
|
err = updateJSONConfigs(v, &jcfg.Informer)
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
return DefaultJSONMarshal(jcfg)
|
|
}
|