概要

GolangでCassandraをセッション管理用のDBとして使ってみたので、記事にしました。Cassandraは、複数台でクラスターを組んで分散DBを作成し、スケールアウトすることが容易にできる構造になっています。また処理性能も構成するノード数に比例します。そのためサーバー管理者としては、比較的安価にスケーラビリティを確保できます。

Cassandraは　テーブルにデータを格納しCQLというSQLのようなクエリ言語を利用してデータのやりとりをすることができるため、直観的にはリレーショナルDBを使っているかのように操作ができます。以前、　【Golang】 mysql データベース処理として、SQLライブラリを用いたGolangによる処理を記事にしましたが、Cassandra DBに関しても、CQLを用いてほとんど同じような処理になります。

※環境は、Macで行っています。
※Golangの基本的な設定や使い方は、はじめてのGolang Webアプリケーション ~ テスト, Dockerコンテナ化までを参考にしてください。

Cassandraの基礎 (インストールとDB, テーブル作成)

以下のコマンドでインストールします。

brew install cassandra
sudo pip install cql

以下のコマンドで、oauthというDBとaccess_tokenというテーブルを作成しています。

describe keyspaces;
>system_traces  system_schema  system_auth  system  system_distributed

CREATE KEYSPACE oauth WITH REPLICATION = {'class':'SimpleStrategy', 'replication_factor':1};

describe keyspaces;
>system_schema  system_auth  system  system_distributed  oauth  system_traces

USE oauth;

CREATE TABLE access_tokens( access_token varchar PRIMARY KEY, user_id bigint, expires bigint);

describe tables;
>access_tokens

SELECT * from access_tokens where access_token='doNaDonaEtc';

 access_token | expires | user_id
--------------+---------+---------

Golang gocql パッケージのインストール

gocqlのパッケージをインストールします。

go get github.com/gocql/gocq

本記事で行う処理例

本記事では、以下の通り処理に対しての例を示す。また、以下に、この処理を行うプログラムを示しています。

access_tokenというデータを作成。
dbRepoにCassandraDBのインターフェースを注入(DB処理のpkgは分離している)
Create methodでデータを挿入
Get methodでデータを取得

※accessTokenは、access_tokenの構造体や処理を集めたパッケージです。本記事の流れとは関係ないため、説明で省略します。

package main

import (
	"fmt"

	access_token "github.com/k-washi/golang-cookbook/cassandraDB/c1/c1/accessToken"

	"github.com/k-washi/golang-cookbook/cassandraDB/c1/c1/oauthdomain"
)

func main() {
	at := access_token.GetAccessToken()
	at.AccessToken = "doNaDonaEtc"
	at.UserID = 123

	dbRepo := oauthdomain.NewRepo()
	err := dbRepo.Create(at)
	if err != nil {
		fmt.Println(err)
	}

	accessToken, err := dbRepo.GetID(at.AccessToken)
	if err != nil {
		fmt.Println(err)
	}
	fmt.Println(accessToken) //&{doNaDonaEtc 123 1580498041}
	/*
			cqlsh:oauth> SELECT * from access_tokens where access_token='doNaDonaEtc';

		 access_token | expires    | user_id
		--------------+------------+---------
			doNaDonaEtc | 1580498041 |     123
	*/
}

Cassandraの設定

Golang でCassandraの設定を行うプログラムを示す。GetSession()によるSessionを、DBの各メソッドにて接続する際に用いる。

package cassandra

import (
	"fmt"
	"github.com/gocql/gocql"
)

var (
	session *gocql.Session
)

func init() {
	// connect to Cassandra
	cluster := gocql.NewCluster("127.0.0.1")
	cluster.Keyspace = "oauth"
	cluster.Consistency = gocql.Quorum

	var err error
	if session, err = cluster.CreateSession(); err != nil {
		panic(err)
	}
	fmt.Println("cassandra config success")
}

func GetSession() *gocql.Session {
	return session
}

Cassandra DBのメソッド

以下にGET, CREATE, UPDATEメソッドを実装しています。queryとして、SQLに似たCQLを書いています。Cassandraの設定を受け取って、それに対して、Queryを送信に、その結果を受け取っています。

package oauthdomain

import (
	"errors"

	access_token "github.com/k-washi/golang-cookbook/cassandraDB/c1/c1/accessToken"

	"github.com/gocql/gocql"

	"github.com/k-washi/golang-cookbook/cassandraDB/c1/c1/cassandra"
)

const (
	queryGetAcessToken     = "SELECT access_token, user_id, expires FROM access_tokens WHERE access_token=?;"
	queryCreateAccessToken = "INSERT INTO access_tokens(access_token, user_id, expires) VALUES (?, ?, ?);"
	queryUpdateExpires     = "UPDATE access_tokens SET expires=? WHERE access_token=?;"
)

func NewRepo() DbRepo {
	return &dbRepo{}
}

type DbRepo interface {
	GetID(string) (*access_token.AccessToken, error)
	Create(access_token.AccessToken) error
	UpdateExpirationTime(access_token.AccessToken) error
}

type dbRepo struct{}

func (r *dbRepo) GetID(at string) (*access_token.AccessToken, error) {
	session := cassandra.GetSession()

	var result access_token.AccessToken
	if err := session.Query(queryGetAcessToken, at).Scan(
		&result.AccessToken,
		&result.UserID,
		&result.Expires,
	); err != nil {
		if err == gocql.ErrNotFound {
			return nil, errors.New("no access token found with given id")
		}
		return nil, err
	}

	return &result, nil

}

func (r *dbRepo) Create(at access_token.AccessToken) error {
	session := cassandra.GetSession()

	if err := session.Query(queryCreateAccessToken,
		at.AccessToken,
		at.UserID,
		at.Expires,
	).Exec(); err != nil {
		return err
	}
	return nil
}

func (r *dbRepo) UpdateExpirationTime(at access_token.AccessToken) error {
	session := cassandra.GetSession()

	if err := session.Query(queryUpdateExpires,
		at.Expires,
		at.AccessToken,
	).Exec(); err != nil {
		return err
	}
	return nil
}