kubernetes

README

Kubernetes Data Source

Provides SQL Access to Kubernetes Rest API's via the DataUX Mysql Proxy Service.

Try It Out

Assumes minikube is running locally

# start dataux service & deployment inside your kube cluster
kubectl create -f https://raw.githubusercontent.com/dataux/dataux/master/backends/kubernetes/conf/dataux.yaml

kubectl describe service dataux

kubectl logs --tail=1000 -lapp=dataux



# connect with mysql client
mysql -h $(minikube service --url --format="{{.IP}}" dataux) -P 30036

show databases;

use kubernetes;

show tables;

describe pods;

select name, creationtimestamp, hostip, podip, hostname from pods;

select name, creationtimestamp, hostip, podip, hostname from pods WHERE name LIKE "dataux%";

select * from nodes;

select count(*) from nodes;

select count(name), hostip from pods GROUP BY hostip;

Install via Download

Install dataux https://github.com/dataux/dataux/releases

minikube start

# install dataux  https://github.com/dataux/dataux/releases
# checkout/clone github.com/dataux/dataux  to get a config file
cd github.com/dataux/dataux

# run 
./dataux --config=backends/kubernetes/kubernetes.conf


# from another terminal

mysql -h 127.0.0.1 -P 4000 -Dkube

Testing & Dev

We need to stand up either a local (minikube) kube cluster or a google container engine cluster. Then we need to decide how to connect to it

• http://kubernetes.io/docs/user-guide/accessing-the-cluster/
• https://github.com/kubernetes/contrib/tree/master/ingress/controllers/nginx

Types & Schema http://kubernetes.io/docs/api-reference/v1/definitions/

minikube start

minikube dashboard

# start a couple of machines
kubectl run hello-minikube --image=gcr.io/google_containers/echoserver:1.4 --port=8080
kubectl expose deployment hello-minikube --type=NodePort

kubectl get pod

curl $(minikube service hello-minikube --url)

kubectl get pods --all-namespaces

kubectl cluster-info


kubectl run dataux --image=gcr.io/dataux-io/dataux:v0.0.1 --port=4000
kubectl expose deployment dataux --type=NodePort

minikube delete --v=10 --show-libmachine-logs --alsologtostderr

Documentation

Overview

package kubernetes implements a data source (backend) to allow dataux to use sql against kubernetes rest/grpc api's

Index

• Constants
• Variables
• type Mutator
• type Object
• type ObjectIterator
  • func Objects(ctx context.Context, kind string, q *SqlToKube) *ObjectIterator
  • func (it *ObjectIterator) Next() (*Object, error)
  • func (it *ObjectIterator) PageInfo() *iterator.PageInfo
• type ResultReader
  • func NewResultReader(req *SqlToKube) *ResultReader
  • func (m *ResultReader) Close() error
  • func (m *ResultReader) Run() error
• type Source
  • func (m *Source) Close() error
  • func (m *Source) DataSource() schema.Source
  • func (m *Source) Init()
  • func (m *Source) Open(tableName string) (schema.Conn, error)
  • func (m *Source) Setup(ss *schema.SchemaSource) error
  • func (m *Source) Table(table string) (*schema.Table, error)
  • func (m *Source) Tables() []string
• type SqlToKube
  • func NewSqlToKube(s *Source, t *schema.Table) (*SqlToKube, error)
  • func (m *SqlToKube) CreateMutator(pc interface{}) (schema.ConnMutator, error)
  • func (m *SqlToKube) Delete(key driver.Value) (int, error)
  • func (m *SqlToKube) DeleteExpression(p interface{}, where expr.Node) (int, error)
  • func (m *SqlToKube) Put(ctx context.Context, key schema.Key, val interface{}) (schema.Key, error)
  • func (m *SqlToKube) PutMulti(ctx context.Context, keys []schema.Key, src interface{}) ([]schema.Key, error)
  • func (m *SqlToKube) WalkExecSource(p *plan.Source) (exec.Task, error)
  • func (m *SqlToKube) WalkSourceSelect(planner plan.Planner, p *plan.Source) (plan.Task, error)

Constants

const (
	DataSourceLabel = "kubernetes"
)

Variables

var (
	// Default page size limit
	DefaultLimit = 5000

	Timeout = 10 * time.Second
)

var (
	ErrNoSchema = fmt.Errorf("No schema or configuration exists")
)

Types

type Mutator

type Mutator struct {
	// contains filtered or unexported fields
}

Mutator a mutator connection

type Object

type Object struct {
	// contains filtered or unexported fields
}

type ObjectIterator

type ObjectIterator struct {
	// contains filtered or unexported fields
}

An ObjectIterator is an iterator over Object which is a generic Object type representing a kube object.

func Objects

func Objects(ctx context.Context, kind string, q *SqlToKube) *ObjectIterator

Objects returns an iterator over the objects in the bucket that match the Query q. If q is nil, no filtering is done.

func (*ObjectIterator) Next

func (it *ObjectIterator) Next() (*Object, error)

Next returns the next result. Its second return value is iterator.Done if there are no more results. Once Next returns iterator.Done, all subsequent calls will return iterator.Done.

func (*ObjectIterator) PageInfo

func (it *ObjectIterator) PageInfo() *iterator.PageInfo

PageInfo supports pagination. See the google.golang.org/api/iterator package for details.

type ResultReader

type ResultReader struct {
	*exec.TaskBase

	Total int
	// contains filtered or unexported fields
}

ResultReader implements result paging, reading of the json/grpc api responses from kube

func NewResultReader

func NewResultReader(req *SqlToKube) *ResultReader

func (*ResultReader) Close

func (m *ResultReader) Close() error

func (*ResultReader) Run

func (m *ResultReader) Run() error

Runs the Kubernetes Result exec

type Source

type Source struct {
	// contains filtered or unexported fields
}

Source is a Kubernetes datasource, this provides Reads, Insert, Update, Delete - singleton shared instance - creates clients to kube api (clients perform queries) - provides schema info about the apis

func (*Source) Close

func (m *Source) Close() error

func (*Source) DataSource

func (m *Source) DataSource() schema.Source

func (*Source) Init

func (m *Source) Init()

func (*Source) Open

func (m *Source) Open(tableName string) (schema.Conn, error)

func (*Source) Setup

func (m *Source) Setup(ss *schema.SchemaSource) error

Setup accepts the schema source config

func (*Source) Table

func (m *Source) Table(table string) (*schema.Table, error)

func (*Source) Tables

func (m *Source) Tables() []string

type SqlToKube

type SqlToKube struct {
	*exec.TaskBase
	// contains filtered or unexported fields
}

SqlToKube Convert a Sql Query to a kubernetes api call - responsible for pushing down as much logic to kube api's as possible - dialect translator

func NewSqlToKube

func NewSqlToKube(s *Source, t *schema.Table) (*SqlToKube, error)

NewSqlToKube create a SQL ast -> Kubernetes Translater

func (*SqlToKube) CreateMutator

func (m *SqlToKube) CreateMutator(pc interface{}) (schema.ConnMutator, error)

CreateMutator part of Mutator interface to allow data sources create a stateful

mutation context for update/delete operations.

func (*SqlToKube) Delete

func (m *SqlToKube) Delete(key driver.Value) (int, error)

Delete delete by row

func (*SqlToKube) DeleteExpression

func (m *SqlToKube) DeleteExpression(p interface{}, where expr.Node) (int, error)

DeleteExpression - delete by expression (where clause)

• For where columns contained in Partition Keys we can push to cassandra
• for others we might have to do a select -> delete

func (*SqlToKube) Put

func (m *SqlToKube) Put(ctx context.Context, key schema.Key, val interface{}) (schema.Key, error)

Put Interface for mutation (insert, update)

func (*SqlToKube) PutMulti

func (m *SqlToKube) PutMulti(ctx context.Context, keys []schema.Key, src interface{}) ([]schema.Key, error)

func (*SqlToKube) WalkExecSource

func (m *SqlToKube) WalkExecSource(p *plan.Source) (exec.Task, error)

WalkExecSource an interface of executor that allows this source to create its own execution Task so that it can push down as much as it can to kubernetes api.

func (*SqlToKube) WalkSourceSelect

func (m *SqlToKube) WalkSourceSelect(planner plan.Planner, p *plan.Source) (plan.Task, error)

WalkSourceSelect An interface implemented by this connection allowing the planner to push down as much logic into this source as possible