Note:
This code influeneced the Go client that's now maintained by Basho, the creators of Riak. You can contribute to the repository here or install with go get github.com/basho/riak-go-client.
Riaken
Go Protocol Buffer driver for the Riak distributed database
Install
Development
go get github.com/riaken/riaken-core
2.0.x Compatible
1.4.x Compatible
go get http://gopkg.in/riaken/riaken-core.v1
Documentation
Development
http://godoc.org/gopkg.in/riaken/riaken-core.v1
2.0.x
1.4.x
http://godoc.org/gopkg.in/riaken/riaken-core.v1
Extended Riaken
There are some modules which wrap/extend/test Riaken located at the following:
- https://github.com/riaken/riaken-struct - Wraps core with higher level struct functionality.
- https://github.com/riaken/riaken-test - Does integration testing against core and struct.
Philosophy
The following points are what drive this project.
- Extendable. riaken-core is the least common demoninator. It is meant to be extended by projects like riaken-struct which add new behavior, such as the ability to convert high level struct data to/from JSON.
- Speed. The current driver clocks at roughly 1800 ops/sec on a single 3.4 GHz Intel Core i7 iMac with 16gb of memory running 5 default Riak instances. This should scale much higher against a real server cluster.
Usage
Client - Basic Initialization
The client is the all encompassing framework of riaken. From there sessions should be grabbed and released as necessary.
package main
import "log"
import "github.com/riaken/riaken-core"
func main() {
// Riak cluster addresses
addrs := []string{"127.0.0.1:8083", "127.0.0.1:8084", "127.0.0.1:8085", "127.0.0.1:8086", "127.0.0.1:8087"}
// Create a client, passing the addresses, and number of connections to maintain per cluster node
client := riaken_core.NewClient(addrs, 1)
// Dial the servers
if err := client.Dial(); err != nil {
log.Fatal(err.Error()) // all nodes are down
}
// Gracefully close all the connections when finished
defer client.Close()
// Grab a session to interact with the cluster
session := client.Session()
// Release the session
defer session.Release()
}
Client Operations
Ping
Simply see if the server is responsive.
if !session.Ping() {
log.Error("no ping response")
}
Set Client ID
if ok, err := session.SetClientId([]byte("client1")); !ok {
log.Error("no response")
} else if err != nil {
log.Error(err.Error())
}
Get Client ID
res, err := session.GetClientId()
if err != nil {
log.Error(err.Error())
}
log.Print(string(res.GetClientId()))
List Buckets
WARNING: This is not recommended to be used against a production server.
buckets, err := session.ListBuckets()
if err != nil {
log.Error(err.Error())
}
Server Info
Get useful info about the Riak servers.
info, err := session.ServerInfo()
if err != nil {
log.Error(err.Error())
}
log.Print(string(info.GetNode()))
log.Print(string(info.GetServerVersion()))
Bucket Operations
Buckets now have a recommended Type() method which allows for another level of namespacing.
bucket := session.GetBucket("bucket").Type("high_level")
The type is set to default by Riak if not specified.
Set Bucket Properties
Not exactly straightforward because they require the use of the RPB structs.
// Make sure to require this
import (
"github.com/riaken/riaken-core/rpb"
"github.com/golang/protobuf/proto"
)
// Code
bucket := session.GetBucket("b2")
props := &rpb.RpbBucketProps{
NVal: proto.Uint32(1),
AllowMult: proto.Bool(true),
}
if ok, err := bucket.SetBucketProps(props); !ok {
log.Error("could not set bucket props")
} else if err != nil {
log.Error(err.Error())
}
Get Bucket Properties
out, err := bucket.GetBucketProps()
if err != nil {
log.Error(err.Error())
}
if out.GetProps().GetAllowMult() != true {
log.Errorf("expected: true, got: %t", out.GetProps().GetAllowMult())
}
Set Bucket Type
bucket := session.GetBucket("b2").Type("test_maps")
props := &rpb.RpbBucketProps{
AllowMult: proto.Bool(true),
}
if ok, err := bucket.SetBucketType(props); !ok {
t.Error("could not set bucket props")
} else if err != nil {
t.Error(err.Error())
}
Reset Bucket
bucket := session.GetBucket("b2").Type("test_maps")
if ok, err := bucket.ResetBucket(); !ok {
t.Error("could not set bucket props")
} else if err != nil {
t.Error(err.Error())
}
List Keys
WARNING: This is not recommended to be used against a production server.
Note that this is a streaming response and must be called until done. If not streamed the next operation will fail and/or hang.
var keys [][]byte
// Loop until done is received from Riak
for out, err := bucket.ListKeys(); !out.GetDone(); out, err = bucket.ListKeys() {
if err != nil {
log.Error(err.Error())
break
}
keys = append(keys, out.GetKeys()...)
}
log.Print(keys)
Object Operations
Store
Grab a bucket by name, an object with the key to store, and insert the data as []byte.
bucket := session.GetBucket("b1")
object := bucket.Object("o1")
if _, err := object.Store([]byte("o1-data")); err != nil {
log.Error(err.Error())
}
For server assigned keys.
bucket := session.GetBucket("b1")
object := bucket.Object("") // leave this blank
res, err := object.Store([]byte("o1-data")) // fetch the response
if err != nil {
log.Error(err.Error())
}
log.Print(string(res.GetKey())) // this is the server assigned key
Fetch
Note that the content comes back as an array. If > 1 this record has siblings.
data, err := object.Fetch()
if err != nil {
log.Error(err.Error())
}
log.Print(string(data.GetContent()[0].GetValue()))
Delete
Verbose version.
if ok, err := object.Delete(); !ok {
log.Error("deletion of object failed")
} else if err != nil {
log.Error(err.Error())
}
Simple version.
if _, err := object.Delete(); err != nil {
log.Error(err.Error())
}
Counter Operations
Update
bucket := session.GetBucket("b5")
counter := bucket.Counter("c1")
if _, err := counter.Update(1); err != nil {
log.Error(err.Error())
}
Get
data, err := counter.Get()
if err != nil {
t.Error(err.Error())
}
log.Print(data.GetValue())
CRDTs
CRDTs can be queried similar to an Object.
crdt := bucket.Crdt("foo")
if _, err := crdt.Fetch(); err != nil {
t.Fatal(err.Error())
}
They will then have their Counter, Set, or Map value set depending on what was initially stored in Riak.
// crdt.Counter
// crdt.Set
// crdt.Map
They are simply deleted with the Object interface.
object := bucket.Object("foo")
if _, err := object.Delete(); err != nil {
t.Fatal(err.Error())
}
Counters
bucket := session.GetBucket("crdt_counter").Type("test_counters")
counter := bucket.Crdt("foo").NewCounter()
counter.Increment(4)
if _, err := counter.Commit(); err != nil {
t.Fatal(err.Error())
}
counter.Decrement(1)
if _, err := counter.Commit(); err != nil {
t.Fatal(err.Error())
}
log.Print(counter.Value) // should equal 3
Sets
bucket := session.GetBucket("crdt_set").Type("test_sets")
set := bucket.Crdt("foo").NewSet()
set.Add("bar")
set.Add("baz")
if _, err := set.Commit(); err != nil {
t.Fatal(err.Error())
}
log.Print(set.Values) // should contain ["bar", "baz"]
set.Remove("baz")
set.Add("car")
if _, err := set.Commit(); err != nil {
t.Fatal(err.Error())
}
log.Print(set.Values) // should contain ["bar", "car"]
Maps
Maps contain:
- Flags
- Registers
- Counters
- Sets
- Maps
They are used in the following way:
bucket := session.GetBucket("crdt_map").Type("test_maps")
crdt := bucket.Crdt("foo")
mp := crdt.NewMap()
// Add Flags
mp.Flags["f1"] = true
mp.Flags["f2"] = false
// Add Registers
mp.Registers["r1"] = "1r"
mp.Registers["r2"] = "2r"
// Add Counters
mp.Counters["c1"] = crdt.NewCounter()
mp.Counters["c1"].Increment(10)
// Add Sets
mp.Sets["s1"] = crdt.NewSet()
mp.Sets["s1"].Add("1")
mp.Sets["s1"].Add("2")
mp.Sets["s1"].Add("3")
mp.Sets["s2"] = crdt.NewSet()
mp.Sets["s2"].Add("a")
mp.Sets["s2"].Add("b")
// Add Maps (within Maps, within...)
mp.Maps["m1"] = crdt.NewMap()
mp.Maps["m1"].Flags["ff1"] = true
mp.Maps["m1"].Registers["rr1"] = "1rr"
mp.Maps["m1"].Counters["cc1"] = crdt.NewCounter()
mp.Maps["m1"].Counters["cc1"].Increment(20)
mp.Maps["m1"].Sets["ss1"] = crdt.NewSet()
mp.Maps["m1"].Sets["ss1"].Add("111")
mp.Maps["m1"].Sets["ss1"].Add("222")
mp.Maps["m1"].Maps["mm1"] = crdt.NewMap()
mp.Maps["m1"].Maps["mm1"].Flags["fff1"] = false
// Save
if _, err := mp.Commit(); err != nil {
t.Fatal(err.Error())
}
Values can be removed from Maps with Remove():
crdt := bucket.Crdt("foo")
if _, err := crdt.Fetch(); err != nil {
t.Fatal(err.Error())
}
crdt.Remove(CRDT_MAP_FLAG, "f2")
if _, err := mp.Commit(); err != nil {
t.Fatal(err.Error())
}
// crdt.Map.Flags["f2"] should no longer exist
Query Operations
Map Reduce
Note that this is a streaming response and must be called until done. If not streamed the next operation will fail and/or hang.
request := []byte(`{
"inputs":"training",
"query":[{"map":{"language":"javascript",
"source":"function(riakObject) {
var val = riakObject.values[0].data.match(/pizza/g);
return [[riakObject.key, (val ? val.length : 0 )]];
}"}}]}`)
contentType := []byte("application/json")
// Query
query := session.Query()
var result []byte
// Loop until done is received from Riak.
for out, err := query.MapReduce(request, contentType); !out.GetDone(); out, err = query.MapReduce(request, contentType) {
if err != nil {
log.Error(err.Error())
break
}
result = append(result, out.GetResponse()...)
}
log.Print(data.GetKeys())
Secondary Indexes
Note that storage_backend must be set to riak_kv_eleveldb_backend in app.config to use this.
query := session.Query()
data, err := query.SecondaryIndexes([]byte("b1"), []byte("animal_bin"), []byte("chicken"), nil, nil, 0, nil)
if err != nil {
log.Error(err.Error())
}
Search
Note that riak_search needs to be enabled in the app.config to use this.
query := session.Query()
data, err := query.Search([]byte("b3"), []byte("food:pizza"))
if err != nil {
log.Error(err.Error())
}
log.Print(data.GetNumFound())
Additional Complex Parameters
Sometimes it is desirable to pass more complex options to the server. All methods capable of receiving additional options have access to Do(). This method takes in a RPB struct and is chained together with the method one wishes to call.
// Make sure to require this
import (
"github.com/riaken/riaken-core/rpb"
"github.com/golang/protobuf/proto"
)
// Code
bucket := session.GetBucket("b1")
object := bucket.Object("o1")
// Store
opts1 := &rpb.RpbPutReq{
ReturnBody: proto.Bool(true),
}
ret, err := object.Do(opts1).Store([]byte("o1-data"))
if err != nil {
log.Error(err.Error())
}
if string(ret.GetContent()[0].GetValue()) != "o1-data" {
log.Errorf("got %s, expected o1-data", string(ret.GetContent()[0].GetValue()))
}
// Fetch
opts2 := &rpb.RpbGetReq{
Head: proto.Bool(true),
}
data, err := object.Do(opts2).Fetch()
if err != nil {
log.Error(err.Error())
}
if len(data.GetContent()) > 0 {
if string(data.GetContent()[0].GetValue()) != "" {
log.Error("expected empty content")
}
}
// Delete
opts3 := &rpb.RpbDelReq{
Rw: proto.Uint32(1),
}
if ok, err := object.Do(opts3).Delete(); !ok {
log.Error("deletion of object failed")
} else if err != nil {
log.Error(err.Error())
}
See the Riak PBC docs for a more detailed explanation of what all the parameters are for each method.
Author
Brian Jones - [email protected] - https://twitter.com/mojobojo