cascade/ipfs-cluster
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Renames of configuration keys and more docs

Browse Source 
License: MIT
Signed-off-by: Hector Sanjuan <hector@protocol.ai>
This commit is contained in:
Hector Sanjuan 2016-12-16 19:14:45 +01:00
parent ba31e35c1a
commit 4ca39ae9b3
9 changed files with 70 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cluster.go
View File

@ -435,7 +435,7 @@ func makeHost(ctx context.Context, cfg *Config) (host.Host, error) {
publicKey := privateKey.GetPublic()
addr, err := multiaddr.NewMultiaddr(fmt.Sprintf("/ip4/%s/tcp/%d",
cfg.ConsensusListenAddr, cfg.ConsensusListenPort))
cfg.ClusterAddr, cfg.ClusterPort))
if err != nil {
return nil, err
}

41
config.go
View File

@ -6,18 +6,35 @@ import (
)
type Config struct {
IPFSHost string `json:"ipfs_host"`
IPFSPort int `json:"ipfs_port"`
APIListenAddr string `json:"cluster_api_listen_addr"`
APIListenPort int `json:"cluster_api_listen_port"`
IPFSAPIListenAddr string `json:"ipfs_api_listen_addr"`
IPFSAPIListenPort int `json:"ipfs_api_listen_port"`
ConsensusListenAddr string `json:"consensus_listen_addr"`
ConsensusListenPort int `json:"consensus_listen_port"`
ClusterPeers []string `json:"cluster_peers"`
ID string `json:"id"`
PrivateKey string `json:"private_key"`
RaftFolder string `json:"raft_folder"`
// Libp2p ID and private key for Cluster communication (including)
// the Consensus component.
ID string `json:"id"`
PrivateKey string `json:"private_key"`
// List of multiaddresses of the peers of this cluster.
ClusterPeers []string `json:"cluster_peers"`
// Listen parameters for the Cluster libp2p Host. Used by
// the Remote RPC and Consensus components.
ClusterAddr string `json:"cluster_addr"`
ClusterPort int `json:"cluster_port"`
// Storage folder for snapshots, log store etc. Used by
// the Consensus component.
ConsensusDataFolder string `json:"consensus_data_folder"`
// Listen parameters for the the Cluster HTTP API component.
APIAddr string `json:"api_addr"`
APIPort int `json:"api_port"`
// Listen parameters for the IPFS Proxy. Used by the IPFS
// connector component.
IPFSAPIAddr string `json:"ipfs_api_addr"`
IPFSAPIPort int `json:"ipfs_api_port"`
// Host/Port for the IPFS daemon.
IPFSAddr string `json:"ipfs_addr"`
IPFSPort int `json:"ipfs_port"`
}
func LoadConfig(path string) (*Config, error) {

2
ipfs-cluster/main.go
View File

@ -40,7 +40,7 @@ func main() {
state := ipfscluster.NewMapState()
tracker := ipfscluster.NewMapPinTracker()
remote := NewLibp2pRemote()
remote := ipfscluster.NewLibp2pRemote()
cluster, err := ipfscluster.NewCluster(clusterCfg, api, proxy, state, tracker, remote)
if err != nil {

8
ipfs_connector.go → ipfs_http_connector.go
View File

@ -49,7 +49,7 @@ type ipfsError struct {
func NewIPFSHTTPConnector(cfg *Config) (*IPFSHTTPConnector, error) {
ctx := context.Background()
l, err := net.Listen("tcp", fmt.Sprintf("%s:%d",
cfg.IPFSAPIListenAddr, cfg.IPFSAPIListenPort))
cfg.IPFSAPIAddr, cfg.IPFSAPIPort))
if err != nil {
return nil, err
}
@ -62,10 +62,10 @@ func NewIPFSHTTPConnector(cfg *Config) (*IPFSHTTPConnector, error) {
ipfs := &IPFSHTTPConnector{
ctx: ctx,
destHost: cfg.IPFSHost,
destHost: cfg.IPFSAddr,
destPort: cfg.IPFSPort,
listenAddr: cfg.IPFSAPIListenAddr,
listenPort: cfg.IPFSAPIListenPort,
listenAddr: cfg.IPFSAPIAddr,
listenPort: cfg.IPFSAPIPort,
handlers: make(map[string]func(http.ResponseWriter, *http.Request)),
rpcCh: make(chan RPC, RPCMaxQueue),
listener: l,

0
ipfs_connector_test.go → ipfs_http_connector_test.go
View File

13
ipfsmock_test.go
View File

@ -5,16 +5,20 @@ import (
"fmt"
"net/http"
"net/http/httptest"
"net/url"
"strconv"
"strings"
cid "github.com/ipfs/go-cid"
)
// This is an ipfs daemon mock which should sustain the functionality used by
// used by ipfscluster.
// ipfscluster.
type ipfsMock struct {
server *httptest.Server
addr string
port int
pinMap *MapState
}
@ -42,6 +46,13 @@ func newIpfsMock() *ipfsMock {
}
ts := httptest.NewServer(http.HandlerFunc(m.handler))
m.server = ts
url, _ := url.Parse(mock.server.URL)
h := strings.Split(url.Host, ":")
i, _ := strconv.Atoi(h[1])
m.port = i
m.addr = h[0]
return m
}

4
raft.go
View File

@ -56,14 +56,14 @@ func makeLibp2pRaft(cfg *Config, host host.Host, state State, op *clusterLogOp)
raftCfg.Logger = nil
}
logger.Debug("creating file snapshot store")
snapshots, err := hashiraft.NewFileSnapshotStore(cfg.RaftFolder, maxSnapshots, nil)
snapshots, err := hashiraft.NewFileSnapshotStore(cfg.ConsensusDataFolder, maxSnapshots, nil)
if err != nil {
logger.Error("creating file snapshot store: ", err)
return nil, nil, nil, err
}
logger.Debug("creating BoltDB log store")
logStore, err := raftboltdb.NewBoltStore(filepath.Join(cfg.RaftFolder, "raft.db"))
logStore, err := raftboltdb.NewBoltStore(filepath.Join(cfg.ConsensusDataFolder, "raft.db"))
if err != nil {
logger.Error("creating bolt store: ", err)
return nil, nil, nil, err

42
http_api.go → rest_api.go
View File

@ -16,9 +16,9 @@ import (
mux "github.com/gorilla/mux"
)
// ClusterHTTPAPI implements a API and aims to provides
// RESTAPI implements an API and aims to provides
// a RESTful HTTP API for Cluster.
type ClusterHTTPAPI struct {
type RESTAPI struct {
ctx context.Context
listenAddr string
listenPort int
@ -70,11 +70,11 @@ type statusResp []statusCidResp
// NewHTTPAPI creates a new object which is ready to be
// started.
func NewHTTPAPI(cfg *Config) (*ClusterHTTPAPI, error) {
func NewHTTPAPI(cfg *Config) (*RESTAPI, error) {
ctx := context.Background()
l, err := net.Listen("tcp", fmt.Sprintf("%s:%d",
cfg.APIListenAddr,
cfg.APIListenPort))
cfg.APIAddr,
cfg.APIPort))
if err != nil {
return nil, err
}
@ -83,10 +83,10 @@ func NewHTTPAPI(cfg *Config) (*ClusterHTTPAPI, error) {
s := &http.Server{Handler: router}
s.SetKeepAlivesEnabled(true) // A reminder that this can be changed
api := &ClusterHTTPAPI{
api := &RESTAPI{
ctx: ctx,
listenAddr: cfg.APIListenAddr,
listenPort: cfg.APIListenPort,
listenAddr: cfg.APIAddr,
listenPort: cfg.APIPort,
listener: l,
server: s,
rpcCh: make(chan RPC, RPCMaxQueue),
@ -106,7 +106,7 @@ func NewHTTPAPI(cfg *Config) (*ClusterHTTPAPI, error) {
return api, nil
}
func (api *ClusterHTTPAPI) routes() []route {
func (api *RESTAPI) routes() []route {
return []route{
route{
"Members",
@ -165,7 +165,7 @@ func (api *ClusterHTTPAPI) routes() []route {
}
}
func (api *ClusterHTTPAPI) run() {
func (api *RESTAPI) run() {
api.wg.Add(1)
go func() {
defer api.wg.Done()
@ -180,7 +180,7 @@ func (api *ClusterHTTPAPI) run() {
}
// Shutdown stops any API listeners.
func (api *ClusterHTTPAPI) Shutdown() error {
func (api *RESTAPI) Shutdown() error {
api.shutdownLock.Lock()
defer api.shutdownLock.Unlock()
@ -202,11 +202,11 @@ func (api *ClusterHTTPAPI) Shutdown() error {
// RpcChan can be used by Cluster to read any
// requests from this component
func (api *ClusterHTTPAPI) RpcChan() <-chan RPC {
func (api *RESTAPI) RpcChan() <-chan RPC {
return api.rpcCh
}
func (api *ClusterHTTPAPI) versionHandler(w http.ResponseWriter, r *http.Request) {
func (api *RESTAPI) versionHandler(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithCancel(api.ctx)
defer cancel()
rRpc := NewRPC(VersionRPC, nil)
@ -217,7 +217,7 @@ func (api *ClusterHTTPAPI) versionHandler(w http.ResponseWriter, r *http.Request
}
}
func (api *ClusterHTTPAPI) memberListHandler(w http.ResponseWriter, r *http.Request) {
func (api *RESTAPI) memberListHandler(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithCancel(api.ctx)
defer cancel()
rRpc := NewRPC(MemberListRPC, nil)
@ -232,7 +232,7 @@ func (api *ClusterHTTPAPI) memberListHandler(w http.ResponseWriter, r *http.Requ
}
}
func (api *ClusterHTTPAPI) pinHandler(w http.ResponseWriter, r *http.Request) {
func (api *RESTAPI) pinHandler(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithCancel(api.ctx)
defer cancel()
@ -245,7 +245,7 @@ func (api *ClusterHTTPAPI) pinHandler(w http.ResponseWriter, r *http.Request) {
}
}
func (api *ClusterHTTPAPI) unpinHandler(w http.ResponseWriter, r *http.Request) {
func (api *RESTAPI) unpinHandler(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithCancel(api.ctx)
defer cancel()
@ -258,7 +258,7 @@ func (api *ClusterHTTPAPI) unpinHandler(w http.ResponseWriter, r *http.Request)
}
}
func (api *ClusterHTTPAPI) pinListHandler(w http.ResponseWriter, r *http.Request) {
func (api *RESTAPI) pinListHandler(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithCancel(api.ctx)
defer cancel()
rRpc := NewRPC(PinListRPC, nil)
@ -270,7 +270,7 @@ func (api *ClusterHTTPAPI) pinListHandler(w http.ResponseWriter, r *http.Request
}
func (api *ClusterHTTPAPI) statusHandler(w http.ResponseWriter, r *http.Request) {
func (api *RESTAPI) statusHandler(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithCancel(api.ctx)
defer cancel()
rRpc := NewRPC(StatusRPC, nil)
@ -280,7 +280,7 @@ func (api *ClusterHTTPAPI) statusHandler(w http.ResponseWriter, r *http.Request)
}
}
func (api *ClusterHTTPAPI) statusCidHandler(w http.ResponseWriter, r *http.Request) {
func (api *RESTAPI) statusCidHandler(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithCancel(api.ctx)
defer cancel()
@ -293,7 +293,7 @@ func (api *ClusterHTTPAPI) statusCidHandler(w http.ResponseWriter, r *http.Reque
}
}
func (api *ClusterHTTPAPI) syncHandler(w http.ResponseWriter, r *http.Request) {
func (api *RESTAPI) syncHandler(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithCancel(api.ctx)
defer cancel()
rRpc := NewRPC(LocalSyncRPC, nil)
@ -303,7 +303,7 @@ func (api *ClusterHTTPAPI) syncHandler(w http.ResponseWriter, r *http.Request) {
}
}
func (api *ClusterHTTPAPI) syncCidHandler(w http.ResponseWriter, r *http.Request) {
func (api *RESTAPI) syncCidHandler(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithCancel(api.ctx)
defer cancel()

0
http_api_test.go → rest_api_test.go
View File