raftproxy.*RaftProxy.OpenRaft - Code Metrics - arazmj/gerdu - Measure and Improve Code Quality continuously with Scrutinizer

raftproxy.*RaftProxy.OpenRaft C
last analyzed 2020-09-13 00:48 UTC

↳ Parent: raftproxy/raft_proxy.go

Complexity

Conditions

Size

Total Lines	70
Code Lines	44

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	10
eloc	44
dl	0
loc	70
rs	5.9999
c	0
b	0
f	0
nop	1

How to fix Long Method Complexity

package raftproxy

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/arazmj/gerdu/cache"
	"github.com/hashicorp/raft"
	raftboltdb "github.com/hashicorp/raft-boltdb"
	log "github.com/sirupsen/logrus"
	"net"
	"net/http"
	"os"
	"path/filepath"
	"time"
)

const (
	raftTimeout         = 10 * time.Second
	retainSnapshotCount = 2
	tcpTimeout          = 10 * time.Second
)

type RaftCache interface {
	raft.FSM
	cache.UnImplementedCache
	OpenRaft(storage string) error
}

type RaftProxy struct {
	raft     *raft.Raft
	Imp      cache.UnImplementedCache
	raftAddr string
	joinAddr string
	localId  string
	RaftCache
}

type command struct {
	Op    string `json:"op,omitempty"`
	Key   string `json:"key,omitempty"`
	Value string `json:"value,omitempty"`
}

func NewRaftProxy(imp cache.UnImplementedCache, raftAddr, joinAddr, localId string) *RaftProxy {
	return &RaftProxy{
		Imp:      imp,
		raftAddr: raftAddr,
		joinAddr: joinAddr,
		localId:  localId,
	}
}

// Put updates or insert a new entry, evicts the old entry
// if cache size is larger than capacity
func (c *RaftProxy) Put(key string, value string) (created bool) {
	cmd := &command{
		Op:    "put",
		Key:   key,
		Value: value,
	}

	future, err := c.applyCommand(cmd)

	if err != nil {
		log.Errorf("Error applyCommand %v", err)
		return false
	}

	if future.Error() != nil {
		log.Errorf("Error in raft apply future %v", future.Error())
		return false
	}

	return future.Response().(bool)
}

func (c *RaftProxy) Delete(key string) (ok bool) {
	cmd := &command{
		Op:  "delete",
		Key: key,
	}

	future, err := c.applyCommand(cmd)

	if err != nil {
		log.Errorf("Error applyCommand %v", err)
		return false
	}

	if future.Error() != nil {
		log.Fatalf("Error in raft apply future %v", future.Error())
		return false
	}

	return future.Response().(bool)
}

func (c *RaftProxy) Get(key string) (value string, ok bool) {
	cmd := &command{
		Op:  "get",
		Key: key,
	}

	future, err := c.applyCommand(cmd)

	if err != nil {
		log.Errorf("Error applyCommand %v", err)
		return "", false
	}

	if future.Error() != nil {
		log.Fatalf("Error in raft apply future %v", future.Error())
		return "", false
	}

	response := future.Response().(getResponse)
	return response.value, response.ok
}

func (c *RaftProxy) applyCommand(cmd *command) (raft.ApplyFuture, error) {
	if c.raft.State() != raft.Leader {
		return nil, errors.New(fmt.Sprintf("not a leader but a %v %p", c.raft.State(), c.raft))
	}

	b, err := json.Marshal(cmd)
	if err != nil {
		return nil, err
	}

	return c.raft.Apply(b, raftTimeout), nil
}

type getResponse struct {
	value string
	ok    bool
}

type fsm RaftProxy

func (f *fsm) Apply(l *raft.Log) interface{} {
	var cmd command
	if err := json.Unmarshal(l.Data, &cmd); err != nil {
		log.Fatalf("failed to unmarshal command: %s", err.Error())
	}

	log.Infof("Apply command: %v", cmd)
	switch cmd.Op {
	case "get":
		value, ok := f.Imp.Get(cmd.Key)
		response := getResponse{
			value: value,
			ok:    ok,
		}
		return response
	case "put":
		return f.Imp.Put(cmd.Key, cmd.Value)
	case "delete":
		return f.Imp.Delete(cmd.Key)
	default:
		log.Fatalf("unrecognized command op: %s", cmd.Op)
	}
	return nil
}

func (c *RaftProxy) OpenRaft(storage string) error {
	// Setup Raft configuration.
	config := raft.DefaultConfig()
	config.LocalID = raft.ServerID(c.localId)

	// Setup Raft communication.
	addr, err := net.ResolveTCPAddr("tcp", c.raftAddr)
	if err != nil {
		return err
	}
	transport, err := raft.NewTCPTransport(c.raftAddr, addr, 3, tcpTimeout, os.Stderr)
	if err != nil {
		return err
	}

	// Create the snapshot store. This allows the Raft to truncate the log.
	snapshots, err := raft.NewFileSnapshotStore(storage, retainSnapshotCount, os.Stderr)
	if err != nil {
		return fmt.Errorf("file snapshot store: %s", err)
	}

	// Create the log store and stable store.
	var logStore raft.LogStore
	var stableStore raft.StableStore
	if storage == "" {
		logStore = raft.NewInmemStore()
		stableStore = raft.NewInmemStore()
	} else {
		boltDB, err := raftboltdb.NewBoltStore(filepath.Join(storage, "raft.db"))
		if err != nil {
			return fmt.Errorf("new bolt store: %s", err)
		}
		logStore = boltDB
		stableStore = boltDB
	}

	// Instantiate the Raft systems.
	ra, err := raft.NewRaft(config, (*fsm)(c), logStore, stableStore, snapshots, transport)
	if err != nil {
		return fmt.Errorf("new raft: %s", err)
	}

	c.raft = ra

	if c.joinAddr == "" {
		configuration := raft.Configuration{
			Servers: []raft.Server{
				{
					ID:      config.LocalID,
					Address: transport.LocalAddr(),
				},
			},
		}
		ra.BootstrapCluster(configuration)
	} else {
		b, err := json.Marshal(map[string]string{"addr": c.raftAddr, "id": c.localId})
		if err != nil {
			return err
		}
		resp, err := http.Post(fmt.Sprintf("http://%s/join", c.joinAddr), "", bytes.NewReader(b))
		if err != nil {
			return err
		}
		defer resp.Body.Close()

		return nil

	}

	return nil
}

// Join joins a node, identified by nodeID and located at addr, to this store.
// The node must be ready to respond to Raft communications at that address.
func (c *RaftProxy) Join(nodeID, addr string) error {
	log.Infof("received join request for remote node %s at %s", nodeID, addr)

	configFuture := c.raft.GetConfiguration()
	if err := configFuture.Error(); err != nil {
		log.Errorf("failed to get raft configuration: %v", err)
		return err
	}

	for _, srv := range configFuture.Configuration().Servers {
		// If a node already exists with either the joining node's ID or address,
		// that node may need to be removed from the config first.
		if srv.ID == raft.ServerID(nodeID) || srv.Address == raft.ServerAddress(addr) {
			// However if *both* the ID and the address are the same, then nothing -- not even
			// a join operation -- is needed.
			if srv.Address == raft.ServerAddress(addr) && srv.ID == raft.ServerID(nodeID) {
				log.Warnf("node %s at %s already member of cluster, ignoring join request", nodeID, addr)
				return nil
			}

			future := c.raft.RemoveServer(srv.ID, 0, 0)
			if err := future.Error(); err != nil {
				return fmt.Errorf("error removing existing node %s at %s: %s", nodeID, addr, err)
			}
		}
	}

	f := c.raft.AddVoter(raft.ServerID(nodeID), raft.ServerAddress(addr), 0, 0)
	if f.Error() != nil {
		return f.Error()
	}
	log.Infof("node %s at %s joined successfully", nodeID, addr)
	return nil
}

func (c *RaftProxy) Leave(nodeID string) error {
	configFuture := c.raft.GetConfiguration()
	if err := configFuture.Error(); err != nil {
		log.Errorf("failed to get raft configuration: %v", err)
		return err
	}

	future := c.raft.RemoveServer(raft.ServerID(nodeID), 0, 0)
	if err := future.Error(); err != nil {
		error := fmt.Sprintf("error removing existing node %s: %s", nodeID, err.Error())
		return errors.New(error)
	}
	return nil
}


1			package raftproxy
2
3			import (
4			"bytes"
5			"encoding/json"
6			"errors"
7			"fmt"
8			"github.com/arazmj/gerdu/cache"
9			"github.com/hashicorp/raft"
10			raftboltdb "github.com/hashicorp/raft-boltdb"
11			log "github.com/sirupsen/logrus"
12			"net"
13			"net/http"
14			"os"
15			"path/filepath"
16			"time"
17			)
18
19			const (
20			raftTimeout = 10 * time.Second
21			retainSnapshotCount = 2
22			tcpTimeout = 10 * time.Second
23			)
24
25			type RaftCache interface {
26			raft.FSM
27			cache.UnImplementedCache
28			OpenRaft(storage string) error
29			}
30
31			type RaftProxy struct {
32			raft *raft.Raft
33			Imp cache.UnImplementedCache
34			raftAddr string
35			joinAddr string
36			localId string
37			RaftCache
38			}
39
40			type command struct {
41			Op string `json:"op,omitempty"`
42			Key string `json:"key,omitempty"`
43			Value string `json:"value,omitempty"`
44			}
45
46			func NewRaftProxy(imp cache.UnImplementedCache, raftAddr, joinAddr, localId string) *RaftProxy {
47			return &RaftProxy{
48			Imp: imp,
49			raftAddr: raftAddr,
50			joinAddr: joinAddr,
51			localId: localId,
52			}
53			}
54
55			// Put updates or insert a new entry, evicts the old entry
56			// if cache size is larger than capacity
57			func (c *RaftProxy) Put(key string, value string) (created bool) {
58			cmd := &command{
59			Op: "put",
60			Key: key,
61			Value: value,
62			}
63
64			future, err := c.applyCommand(cmd)
65
66			if err != nil {
67			log.Errorf("Error applyCommand %v", err)
68			return false
69			}
70
71			if future.Error() != nil {
72			log.Errorf("Error in raft apply future %v", future.Error())
73			return false
74			}
75
76			return future.Response().(bool)
77			}
78
79			func (c *RaftProxy) Delete(key string) (ok bool) {
80			cmd := &command{
81			Op: "delete",
82			Key: key,
83			}
84
85			future, err := c.applyCommand(cmd)
86
87			if err != nil {
88			log.Errorf("Error applyCommand %v", err)
89			return false
90			}
91
92			if future.Error() != nil {
93			log.Fatalf("Error in raft apply future %v", future.Error())
94			return false
95			}
96
97			return future.Response().(bool)
98			}
99
100			func (c *RaftProxy) Get(key string) (value string, ok bool) {
101			cmd := &command{
102			Op: "get",
103			Key: key,
104			}
105
106			future, err := c.applyCommand(cmd)
107
108			if err != nil {
109			log.Errorf("Error applyCommand %v", err)
110			return "", false
111			}
112
113			if future.Error() != nil {
114			log.Fatalf("Error in raft apply future %v", future.Error())
115			return "", false
116			}
117
118			response := future.Response().(getResponse)
119			return response.value, response.ok
120			}
121
122			func (c RaftProxy) applyCommand(cmd command) (raft.ApplyFuture, error) {
123			if c.raft.State() != raft.Leader {
124			return nil, errors.New(fmt.Sprintf("not a leader but a %v %p", c.raft.State(), c.raft))
125			}
126
127			b, err := json.Marshal(cmd)
128			if err != nil {
129			return nil, err
130			}
131
132			return c.raft.Apply(b, raftTimeout), nil
133			}
134
135			type getResponse struct {
136			value string
137			ok bool
138			}
139
140			type fsm RaftProxy
141
142			func (f fsm) Apply(l raft.Log) interface{} {
143			var cmd command
144			if err := json.Unmarshal(l.Data, &cmd); err != nil {
145			log.Fatalf("failed to unmarshal command: %s", err.Error())
146			}
147
148			log.Infof("Apply command: %v", cmd)
149			switch cmd.Op {
150			case "get":
151			value, ok := f.Imp.Get(cmd.Key)
152			response := getResponse{
153			value: value,
154			ok: ok,
155			}
156			return response
157			case "put":
158			return f.Imp.Put(cmd.Key, cmd.Value)
159			case "delete":
160			return f.Imp.Delete(cmd.Key)
161			default:
162			log.Fatalf("unrecognized command op: %s", cmd.Op)
163			}
164			return nil
165			}
166
167			func (c *RaftProxy) OpenRaft(storage string) error {
168			// Setup Raft configuration.
169			config := raft.DefaultConfig()
170			config.LocalID = raft.ServerID(c.localId)
171
172			// Setup Raft communication.
173			addr, err := net.ResolveTCPAddr("tcp", c.raftAddr)
174			if err != nil {
175			return err
176			}
177			transport, err := raft.NewTCPTransport(c.raftAddr, addr, 3, tcpTimeout, os.Stderr)
178			if err != nil {
179			return err
180			}
181
182			// Create the snapshot store. This allows the Raft to truncate the log.
183			snapshots, err := raft.NewFileSnapshotStore(storage, retainSnapshotCount, os.Stderr)
184			if err != nil {
185			return fmt.Errorf("file snapshot store: %s", err)
186			}
187
188			// Create the log store and stable store.
189			var logStore raft.LogStore
190			var stableStore raft.StableStore
191			if storage == "" {
192			logStore = raft.NewInmemStore()
193			stableStore = raft.NewInmemStore()
194			} else {
195			boltDB, err := raftboltdb.NewBoltStore(filepath.Join(storage, "raft.db"))
196			if err != nil {
197			return fmt.Errorf("new bolt store: %s", err)
198			}
199			logStore = boltDB
200			stableStore = boltDB
201			}
202
203			// Instantiate the Raft systems.
204			ra, err := raft.NewRaft(config, (*fsm)(c), logStore, stableStore, snapshots, transport)
205			if err != nil {
206			return fmt.Errorf("new raft: %s", err)
207			}
208
209			c.raft = ra
210
211			if c.joinAddr == "" {
212			configuration := raft.Configuration{
213			Servers: []raft.Server{
214			{
215			ID: config.LocalID,
216			Address: transport.LocalAddr(),
217			},
218			},
219			}
220			ra.BootstrapCluster(configuration)
221			} else {
222			b, err := json.Marshal(map[string]string{"addr": c.raftAddr, "id": c.localId})
223			if err != nil {
224			return err
225			}
226			resp, err := http.Post(fmt.Sprintf("http://%s/join", c.joinAddr), "", bytes.NewReader(b))
227			if err != nil {
228			return err
229			}
230			defer resp.Body.Close()
231
232			return nil
233
234			}
235
236			return nil
237			}
238
239			// Join joins a node, identified by nodeID and located at addr, to this store.
240			// The node must be ready to respond to Raft communications at that address.
241			func (c *RaftProxy) Join(nodeID, addr string) error {
242			log.Infof("received join request for remote node %s at %s", nodeID, addr)
243
244			configFuture := c.raft.GetConfiguration()
245			if err := configFuture.Error(); err != nil {
246			log.Errorf("failed to get raft configuration: %v", err)
247			return err
248			}
249
250			for _, srv := range configFuture.Configuration().Servers {
251			// If a node already exists with either the joining node's ID or address,
252			// that node may need to be removed from the config first.
253			if srv.ID == raft.ServerID(nodeID) \|\| srv.Address == raft.ServerAddress(addr) {
254			// However if both the ID and the address are the same, then nothing -- not even
255			// a join operation -- is needed.
256			if srv.Address == raft.ServerAddress(addr) && srv.ID == raft.ServerID(nodeID) {
257			log.Warnf("node %s at %s already member of cluster, ignoring join request", nodeID, addr)
258			return nil
259			}
260
261			future := c.raft.RemoveServer(srv.ID, 0, 0)
262			if err := future.Error(); err != nil {
263			return fmt.Errorf("error removing existing node %s at %s: %s", nodeID, addr, err)
264			}
265			}
266			}
267
268			f := c.raft.AddVoter(raft.ServerID(nodeID), raft.ServerAddress(addr), 0, 0)
269			if f.Error() != nil {
270			return f.Error()
271			}
272			log.Infof("node %s at %s joined successfully", nodeID, addr)
273			return nil
274			}
275
276			func (c *RaftProxy) Leave(nodeID string) error {
277			configFuture := c.raft.GetConfiguration()
278			if err := configFuture.Error(); err != nil {
279			log.Errorf("failed to get raft configuration: %v", err)
280			return err
281			}
282
283			future := c.raft.RemoveServer(raft.ServerID(nodeID), 0, 0)
284			if err := future.Error(); err != nil {
285			error := fmt.Sprintf("error removing existing node %s: %s", nodeID, err.Error())
286			return errors.New(error)
287			}
288			return nil
289			}
290

arazmj / gerdu

GitHub Access Token became invalid

raftproxy.*RaftProxy.OpenRaft C last analyzed 2020-09-13 00:48 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

raftproxy.*RaftProxy.OpenRaft C
last analyzed 2020-09-13 00:48 UTC