common

package
v0.0.0-...-705a261 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 16, 2015 License: Apache-2.0 Imports: 28 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	INDEX_ADMIN_SERVICE = "indexAdmin"
	INDEX_SCAN_SERVICE  = "indexScan"
	INDEX_HTTP_SERVICE  = "indexHttp"
)
View Source 
const (
	KEY    PartitionScheme = "KEY"
	HASH                   = "HASH"
	RANGE                  = "RANGE"
	TEST                   = "TEST"
	SINGLE                 = "SINGLE"
)
View Source 
const (
	ForestDB = "forestdb"
	MemDB    = "memdb"
)
View Source 
const (
	PayloadKeyVersions byte = iota + 1
	PayloadVbmap
)

types of payload

View Source 
const (
	Upsert         byte = iota + 1 // data command
	Deletion                       // data command
	UpsertDeletion                 // data command
	Sync                           // control command
	DropData                       // control command
	StreamBegin                    // control command
	StreamEnd                      // control command
	Snapshot                       // control command
)

List of possible mutation commands. Mutation messages are broadly divided into data and control messages. The division is based on the command field.

View Source 
const (
	IndexingMetaDir          = "/indexing/"
	IndexingSettingsMetaDir  = IndexingMetaDir + "settings/"
	IndexingSettingsMetaPath = IndexingSettingsMetaDir + "config"
)
View Source 
const BUCKET_UUID_NIL = ""
View Source 
const CLUSTER_INFO_INIT_RETRIES = 5
View Source 
const INDEXER_ID_NIL = IndexerId("")

Variables

View Source 
var (
	ErrInvalidNodeId       = errors.New("Invalid NodeId")
	ErrInvalidService      = errors.New("Invalid service")
	ErrNodeNotBucketMember = errors.New("Node is not a member of bucket")
)
View Source 
var (
	ErrNodeServicesConnect = errors.New("Internal services API connection closed")
	ErrNodeServicesCancel  = errors.New("Cancelled services change notifier")
	ErrNotifierInvalid     = errors.New("Notifier invalidated due to internal error")
)
View Source 
var ErrorChannelFull = errors.New("secondary.channelFull")

ErrorChannelFull

View Source 
var ErrorClosed = errors.New("genServer.closed")

ErrorClosed

View Source 
var ErrorEmptyN1QLExpression = errors.New("secondary.emptyN1QLExpression")

ErrorEmptyN1QLExpression

View Source 
var ErrorInvalidRequest = errors.New("secondary.invalidRequest")

ErrorInvalidRequest

View Source 
var ErrorNotFound = errors.New("secondary.notFound")

ErrorNotFound

View Source 
var ErrorNotMyVbucket = errors.New("secondary.notMyVbucket")

ErrorNotMyVbucket

View Source 
var ErrorUnexpectedPayload = errors.New("secondary.unexpectedPayload")

ErrorUnexpectedPayload

View Source 
var NUM_VBUCKETS int
View Source 
var ProtobufDataPathMajorNum byte // = 0

ProtobufDataPathMajorNum major version number for mutation data path.

View Source 
var ProtobufDataPathMinorNum byte = 1

ProtobufDataPathMinorNum minor version number for mutation data path.

View Source 
var SystemConfig = Config{

	"maxVbuckets": ConfigValue{
		1024,
		"number of vbuckets configured in KV",
		1024,
		true,
	},

	"projector.name": ConfigValue{
		"projector",
		"human readable name for this projector",
		"projector",
		true,
	},
	"projector.clusterAddr": ConfigValue{
		"localhost:9000",
		"KV cluster's address to be used by projector",
		"localhost:9000",
		true,
	},
	"projector.maxCpuPercent": ConfigValue{
		projector_maxCpuPercent,
		"Maximum percent of CPU that projector can use. " +
			"EG, 200% in 4-core (400%) machine would set indexer to " +
			"use 2 cores",
		projector_maxCpuPercent,
		false,
	},

	"projector.routerEndpointFactory": ConfigValue{
		RouterEndpointFactory(nil),
		"RouterEndpointFactory callback to generate endpoint instances " +
			"to push data to downstream",
		RouterEndpointFactory(nil),
		true,
	},
	"projector.feedWaitStreamReqTimeout": ConfigValue{
		10 * 1000,
		"timeout, in milliseconds, to await a response for StreamRequest",
		10 * 1000,
		false,
	},
	"projector.feedWaitStreamEndTimeout": ConfigValue{
		10 * 1000,
		"timeout, in milliseconds, to await a response for StreamEnd",
		10 * 1000,
		false,
	},
	"projector.mutationChanSize": ConfigValue{
		100,
		"channel size of projector's vbucket workers, " +
			"changing this value does not affect existing feeds.",
		100,
		false,
	},
	"projector.feedChanSize": ConfigValue{
		100,
		"channel size for feed's control path, " +
			"changing this value does not affect existing feeds.",
		100,
		false,
	},
	"projector.backChanSize": ConfigValue{
		10000,
		"channel size of projector feed's back-channel, " +
			"changing this value does not affect existing feeds.",
		10000,
		false,
	},
	"projector.syncTimeout": ConfigValue{
		2000,
		"timeout, in milliseconds, for sending periodic Sync messages, " +
			"changing this value does not affect existing feeds.",
		2000,
		false,
	},
	"projector.watchInterval": ConfigValue{
		5 * 60 * 1000,
		"periodic tick, in milli-seconds to check for stale feeds, " +
			"a feed is considered stale when all its endpoint go stale.",
		5 * 60 * 1000,
		true,
	},
	"projector.staleTimeout": ConfigValue{
		5 * 60 * 1000,
		"timeout, in milli-seconds to wait for response for feed's genserver" +
			"feed will be force-shutdown if timeout expires",
		5 * 60 * 1000,
		true,
	},
	"projector.cpuProfFname": ConfigValue{
		"",
		"filename to dump cpu-profile for projector.",
		"",
		false,
	},
	"projector.cpuProfile": ConfigValue{
		false,
		"boolean indicate whether to start or stop projector cpu profiling.",
		false,
		false,
	},
	"projector.memProfFname": ConfigValue{
		"",
		"filename to dump mem-profile for projector.",
		"",
		false,
	},
	"projector.memProfile": ConfigValue{
		false,
		"boolean to take current mem profile from projector.",
		false,
		false,
	},

	"projector.dcp.genChanSize": ConfigValue{
		2048,
		"channel size for DCP's gen-server routine, " +
			"changing this value does not affect existing feeds.",
		2048,
		false,
	},
	"projector.dcp.dataChanSize": ConfigValue{
		10000,
		"channel size for DCP's data path routines, " +
			"changing this value does not affect existing feeds.",
		10000,
		false,
	},

	"projector.adminport.name": ConfigValue{
		"projector.adminport",
		"human readable name for this adminport, must be supplied",
		"projector.adminport",
		true,
	},
	"projector.adminport.listenAddr": ConfigValue{
		"",
		"projector's adminport address listen for request.",
		"",
		true,
	},
	"projector.adminport.urlPrefix": ConfigValue{
		"/adminport/",
		"url prefix (script-path) for adminport used by projector",
		"/adminport/",
		true,
	},
	"projector.adminport.readTimeout": ConfigValue{
		0,
		"timeout in milliseconds, is http server's read timeout, " +
			"also refer to projector.dataport.harakiriTimeout and " +
			"indexer.dataport.tcpReadDeadline",
		0,
		true,
	},
	"projector.adminport.writeTimeout": ConfigValue{
		0,
		"timeout in milliseconds, is http server's write timeout",
		0,
		true,
	},
	"projector.adminport.maxHeaderBytes": ConfigValue{
		1 << 20,
		"in bytes, is max. length of adminport http header",
		1 << 20,
		true,
	},

	"projector.dataport.remoteBlock": ConfigValue{
		true,
		"should dataport endpoint block when remote is slow, " +
			"does not affect existing feeds.",
		true,
		false,
	},
	"projector.dataport.keyChanSize": ConfigValue{
		100000,
		"channel size of dataport endpoints data input, " +
			"does not affect existing feeds.",
		100000,
		true,
	},
	"projector.dataport.bufferSize": ConfigValue{
		100,
		"number of entries to buffer before flushing it, where each entry " +
			"is for a vbucket's set of mutations that was flushed, " +
			"by the endpoint, does not affect existing feeds.",
		100,
		false,
	},
	"projector.dataport.bufferTimeout": ConfigValue{
		25,
		"timeout in milliseconds, to flush vbucket-mutations from, " +
			"endpoint, does not affect existing feeds.",
		25,
		false,
	},
	"projector.dataport.harakiriTimeout": ConfigValue{
		30 * 1000,
		"timeout in milliseconds, after which endpoint will commit harakiri " +
			"if not active, does not affect existing feeds, " +
			"also refer to projector.adminport.readTimeout and " +
			"indexer.dataport.tcpReadDeadline.",
		30 * 1000,
		false,
	},
	"projector.dataport.maxPayload": ConfigValue{
		1024 * 1024,
		"maximum payload length, in bytes, for transmission data from " +
			"router to downstream client, does not affect eixting feeds.",
		1024 * 1024,
		true,
	},

	"manager.projectorclient.retryInterval": ConfigValue{
		16,
		"retryInterval, in milliseconds when connection refused by server",
		16,
		true,
	},
	"manager.projectorclient.maxRetries": ConfigValue{
		5,
		"maximum number of times to retry",
		5,
		true,
	},
	"manager.projectorclient.exponentialBackoff": ConfigValue{
		2,
		"multiplying factor on retryInterval for every attempt with server",
		2,
		true,
	},
	"manager.projectorclient.urlPrefix": ConfigValue{
		"/adminport/",
		"url prefix (script-path) for adminport used by projector",
		"/adminport/",
		true,
	},

	"indexer.dataport.genServerChanSize": ConfigValue{
		10000,
		"request channel size of indexer dataport's gen-server routine",
		10000,
		true,
	},
	"indexer.dataport.maxPayload": ConfigValue{
		1000 * 1024,
		"maximum payload length, in bytes, for receiving data from router",
		1000 * 1024,
		true,
	},
	"indexer.dataport.tcpReadDeadline": ConfigValue{
		30 * 1000,
		"timeout, in milliseconds, while reading from socket, " +
			"also refer to projector.adminport.readTimeout and " +
			"projector.dataport.harakiriTimeout.",
		30 * 1000,
		true,
	},

	"indexer.queryport.maxPayload": ConfigValue{
		1000 * 1024,
		"maximum payload, in bytes, for receiving data from client",
		1000 * 1024,
		true,
	},
	"indexer.queryport.readDeadline": ConfigValue{
		4000,
		"timeout, in milliseconds, is timeout while reading from socket",
		4000,
		true,
	},
	"indexer.queryport.writeDeadline": ConfigValue{
		4000,
		"timeout, in milliseconds, is timeout while writing to socket",
		4000,
		true,
	},
	"indexer.queryport.pageSize": ConfigValue{
		1,
		"number of index-entries that shall be returned as single payload",
		1,
		true,
	},
	"indexer.queryport.streamChanSize": ConfigValue{
		16,
		"size of the buffered channels used to stream request and response.",
		16,
		true,
	},

	"queryport.client.maxPayload": ConfigValue{
		1000 * 1024,
		"maximum payload, in bytes, for receiving data from server",
		1000 * 1024,
		true,
	},
	"queryport.client.readDeadline": ConfigValue{
		300000,
		"timeout, in milliseconds, is timeout while reading from socket",
		300000,
		true,
	},
	"queryport.client.writeDeadline": ConfigValue{
		4000,
		"timeout, in milliseconds, is timeout while writing to socket",
		4000,
		true,
	},
	"queryport.client.settings.poolSize": ConfigValue{
		1000,
		"number simultaneous active connections connections in a pool",
		1000,
		true,
	},
	"queryport.client.settings.poolOverflow": ConfigValue{
		30,
		"maximum number of connections in a pool",
		30,
		true,
	},
	"queryport.client.connPoolTimeout": ConfigValue{
		1000,
		"timeout, in milliseconds, is timeout for retrieving a connection " +
			"from the pool",
		1000,
		true,
	},
	"queryport.client.connPoolAvailWaitTimeout": ConfigValue{
		1,
		"timeout, in milliseconds, to wait for an existing connection " +
			"from the pool before considering the creation of a new one",
		1,
		true,
	},
	"queryport.client.retryScanPort": ConfigValue{
		2,
		"number of times to retry when scanport is not detectable",
		2,
		true,
	},
	"queryport.client.retryIntervalScanport": ConfigValue{
		10,
		"wait, in milliseconds, before re-trying for a scanport",
		10,
		true,
	},
	"queryport.client.servicesNotifierRetryTm": ConfigValue{
		1000,
		"wait, in milliseconds, before restarting the ServicesNotifier",
		1000,
		true,
	},

	"indexer.projectorclient.retryInterval": ConfigValue{
		16,
		"retryInterval, in milliseconds when connection refused by server",
		16,
		true,
	},
	"indexer.projectorclient.maxRetries": ConfigValue{
		5,
		"maximum number of times to retry",
		5,
		true,
	},
	"indexer.projectorclient.exponentialBackoff": ConfigValue{
		2,
		"multiplying factor on retryInterval for every attempt with server",
		2,
		true,
	},
	"indexer.projectorclient.urlPrefix": ConfigValue{
		"/adminport/",
		"url prefix (script-path) for adminport used by projector",
		"/adminport/",
		true,
	},
	"indexer.adminPort": ConfigValue{
		"9100",
		"port for index ddl and status operations",
		"9100",
		true,
	},
	"indexer.scanPort": ConfigValue{
		"9101",
		"port for index scan operations",
		"9101",
		true,
	},
	"indexer.httpPort": ConfigValue{
		"9102",
		"port for external stats amd settings",
		"9102",
		true,
	},
	"indexer.streamInitPort": ConfigValue{
		"9103",
		"port for inital build stream",
		"9103",
		true,
	},
	"indexer.streamCatchupPort": ConfigValue{
		"9104",
		"port for catchup stream",
		"9104",
		true,
	},
	"indexer.streamMaintPort": ConfigValue{
		"9105",
		"port for maintenance stream",
		"9105",
		true,
	},
	"indexer.clusterAddr": ConfigValue{
		"127.0.0.1:8091",
		"Local cluster manager address",
		"127.0.0.1:8091",
		true,
	},
	"indexer.numVbuckets": ConfigValue{
		1024,
		"Number of vbuckets",
		1024,
		true,
	},
	"indexer.enableManager": ConfigValue{
		false,
		"Enable index manager",
		false,
		true,
	},
	"indexer.storage_dir": ConfigValue{
		"./",
		"Index file storage directory",
		"./",
		true,
	},
	"indexer.numSliceWriters": ConfigValue{
		1,
		"Number of Writer Threads for a Slice",
		1,
		true,
	},

	"indexer.sync_period": ConfigValue{
		uint64(2000),
		"Stream message sync interval in millis",
		uint64(2000),
		true,
	},

	"indexer.stats_cache_timeout": ConfigValue{
		uint64(5000),
		"Stats cache ttl in millis",
		uint64(5000),
		true,
	},

	"indexer.settings.compaction.check_period": ConfigValue{
		30,
		"Compaction poll interval in seconds",
		30,
		false,
	},
	"indexer.settings.compaction.interval": ConfigValue{
		"00:00,00:00",
		"Compaction allowed interval",
		"00:00,00:00",
		false,
	},
	"indexer.settings.compaction.min_frag": ConfigValue{
		30,
		"Compaction fragmentation threshold percentage",
		30,
		false,
	},
	"indexer.settings.compaction.min_size": ConfigValue{
		uint64(1024 * 1024),
		"Compaction min file size",
		uint64(1024 * 1024),
		false,
	},
	"indexer.settings.persisted_snapshot.interval": ConfigValue{
		uint64(5000),
		"Persisted snapshotting interval in milliseconds",
		uint64(5000),
		false,
	},
	"indexer.settings.inmemory_snapshot.interval": ConfigValue{
		uint64(200),
		"InMemory snapshotting interval in milliseconds",
		uint64(200),
		false,
	},
	"indexer.settings.recovery.max_rollbacks": ConfigValue{
		5,
		"Maximum number of committed rollback points",
		5,
		false,
	},
	"indexer.settings.memory_quota": ConfigValue{
		uint64(256 * 1024 * 1024),
		"Maximum memory used by the indexer buffercache",
		uint64(256 * 1024 * 1024),
		false,
	},
	"indexer.settings.max_cpu_percent": ConfigValue{
		400,
		"Maximum percent of CPU that indexer can use. " +
			"EG, 200% in 4-core (400%) machine would set indexer to " +
			"use 2 cores",
		400,
		false,
	},
	"indexer.settings.log_level": ConfigValue{
		"debug",
		"Indexer logging level",
		"debug",
		false,
	},
	"indexer.settings.scan_timeout": ConfigValue{
		120000,
		"timeout, in milliseconds, timeout for index scan processing",
		120000,
		true,
	},

	"indexer.settings.send_buffer_size": ConfigValue{
		1024,
		"Buffer size for batching rows during scan result streaming",
		1024,
		true,
	},

	"indexer.settings.cpuProfFname": ConfigValue{
		"",
		"filename to dump cpu-profile for indexer.",
		"",
		false,
	},
	"indexer.settings.cpuProfile": ConfigValue{
		false,
		"boolean indicate whether to start or stop indexer cpu profiling.",
		false,
		false,
	},
	"indexer.settings.memProfFname": ConfigValue{
		"",
		"filename to dump mem-profile for indexer.",
		"",
		false,
	},
	"indexer.settings.memProfile": ConfigValue{
		false,
		"boolean to take current mem profile from indexer.",
		false,
		false,
	},

	"indexer.settings.maxVbQueueLength": ConfigValue{
		uint64(0),
		"Maximum Length of Mutation Queue Per Vbucket. This " +
			"allocation is done per bucket.",
		uint64(10000),
		false,
	},

	"indexer.settings.largeSnapshotThreshold": ConfigValue{
		uint64(200),
		"Threshold For Considering a DCP Snapshot as Large. " +
			"Must be less than maxVbQueueLength.",
		uint64(200),
		false,
	},

	"indexer.settings.sliceBufSize": ConfigValue{
		uint64(50000),
		"Buffer for each slice to queue mutations before flush " +
			"to storage.",
		uint64(50000),
		false,
	},
	"indexer.settings.bufferPoolBlockSize": ConfigValue{
		16 * 1024,
		"Size of memory block in memory pool",
		16 * 1024,
		false,
	},
	"indexer.settings.statsLogDumpInterval": ConfigValue{
		uint64(60),
		"Periodic stats dump logging interval in seconds",
		uint64(60),
		false,
	},
	"indexer.settings.max_writer_lock_prob": ConfigValue{
		20,
		"Controls the write rate for compaction to catch up",
		20,
		false,
	},
	"projector.settings.log_level": ConfigValue{
		"debug",
		"Projector logging level",
		"debug",
		false,
	},
}

SystemConfig is default configuration for system and components. configuration parameters follow flat namespacing like, 

"maxVbuckets"  for system-level config parameter
"projector.xxx" for projector component.
"projector.adminport.xxx" for adminport under projector component.

etc...

Functions

func BucketSeqnos 

func BucketSeqnos(cluster, pooln, bucketn string) (l_seqnos []uint64, err error)

BucketSeqnos return list of {{vbno,seqno}..} for all vbuckets. this call might fail due to,

  • concurrent access that can preserve a deleted/failed bucket object.
  • pollForDeletedBuckets() did not get a chance to cleanup a deleted bucket.

in both the cases if the call is retried it should get fixed, provided a valid bucket exists.

func BucketTs 

func BucketTs(bucket *couchbase.Bucket, maxvb int) (seqnos, vbuuids []uint64, err error)

BucketTs return bucket timestamp for all vbucket.

func ClusterAuthUrl 

func ClusterAuthUrl(cluster string) (string, error)

func ClusterUrl 

func ClusterUrl(cluster string) string

func CommonStrings 

func CommonStrings(xs []string, ys []string) []string

CommonStrings returns intersection of two set of strings.

func ConnectBucket 

func ConnectBucket(cluster, pooln, bucketn string) (*couchbase.Bucket, error)

ConnectBucket will instantiate a couchbase-bucket instance with cluster. caller's responsibility to close the bucket.

func CrashOnError 

func CrashOnError(err error)

func EquivalentIP 

func EquivalentIP(
	raddr string,
	raddrs []string) (this string, other string, err error)

func ExcludeStrings 

func ExcludeStrings(strs []string, excludes []string) []string

ExcludeStrings will exclude strings in `excludes` from `strs`. preserves the order of `strs` in the result.

func ExcludeUint32 

func ExcludeUint32(xs []uint32, from []uint32) []uint32

ExcludeUint32 remove items from list.

func ExcludeUint64 

func ExcludeUint64(xs []uint64, from []uint64) []uint64

ExcludeUint64 remove items from list.

func ExitOnStdinClose 

func ExitOnStdinClose()

ExitOnStdinClose is exit handler to be used with ns-server.

func FailsafeOp 

func FailsafeOp(
	reqch, respch chan []interface{},
	cmd []interface{},
	finch chan bool) ([]interface{}, error)

FailsafeOp can be used by gen-server implementors to avoid infinitely blocked API calls.

func FailsafeOpAsync 

func FailsafeOpAsync(
	reqch chan []interface{}, cmd []interface{}, finch chan bool) error

FailsafeOpAsync is same as FailsafeOp that can be used for asynchronous operation, that is, caller does not wait for response.

func FailsafeOpNoblock 

func FailsafeOpNoblock(
	reqch chan []interface{}, cmd []interface{}, finch chan bool) error

FailsafeOpNoblock is same as FailsafeOpAsync that can be used for non-blocking operation, that is, if `reqch` is full caller does not block.

func FileSize 

func FileSize(name string) (int64, error)

func GetBucketUUID 

func GetBucketUUID(cluster, bucket string) (string, error)

This method fetch the bucket UUID. If this method return an error, then it means that the node is not able to connect in order to fetch bucket UUID.

func GetColocatedHost 

func GetColocatedHost(cluster string) (string, error)

GetColocatedHost find the server addr for localhost and return the same.

func GetKVAddrs 

func GetKVAddrs(cluster, pooln, bucketn string) ([]string, error)

GetKVAddrs gather the list of kvnode-address based on the latest vbmap.

func GetLocalIP 

func GetLocalIP() (net.IP, error)

GetLocalIP return the first external-IP4 configured for the first interface connected to this node.

func HasString 

func HasString(str string, strs []string) bool

HasString does membership check for a string.

func HasUint32 

func HasUint32(item uint32, xs []uint32) bool

HasUint32 does membership check for a uint32 integer.

func HasUint64 

func HasUint64(item uint64, xs []uint64) bool

HasUint64 does membership check for a uint32 integer.

func HashVbuuid 

func HashVbuuid(vbuuids []uint64) uint64

HashVbuuid return crc64 value of list of 64-bit vbuuids.

func IndexStatement 

func IndexStatement(def IndexDefn) string

func Intersection 

func Intersection(this, other []uint16) []uint16

Intersection operation on two sets of vbuckets, return a sorted list of vbuckets present in both set.

func IsAuthValid 

func IsAuthValid(r *http.Request, server string) (bool, error)

func IsIPLocal 

func IsIPLocal(ip string) bool

IsIPLocal return whether `ip` address is loopback address or compares equal with local-IP-address.

func LogOs 

func LogOs() string

func LogRuntime 

func LogRuntime() string

func MarshallIndexDefn 

func MarshallIndexDefn(defn *IndexDefn) ([]byte, error)

func MaxVbuckets 

func MaxVbuckets(bucket *couchbase.Bucket) (int, error)

MaxVbuckets return the number of vbuckets in bucket.

func MaybeSetEnv 

func MaybeSetEnv(key, value string) string

func OpError 

func OpError(err error, vals []interface{}, idx int) error

OpError suppliments FailsafeOp used by gen-servers.

func RemoveString 

func RemoveString(item string, xs []string) []string

RemoveString delete `item` from list `xs`.

func RemoveUint16 

func RemoveUint16(item uint16, xs []uint16) []uint16

RemoveUint16 delete `item` from list `xs`.

func RemoveUint32 

func RemoveUint32(item uint32, xs []uint32) []uint32

RemoveUint32 delete `item` from list `xs`.

func SeedProcess 

func SeedProcess()

func SetNumCPUs 

func SetNumCPUs(percent int) int

func SetupSettingsNotifier 

func SetupSettingsNotifier(callb func(Config), cancelCh chan struct{})

func StreamID 

func StreamID(bucket string, vbno uint16) string

StreamID is unique id for a vbucket across buckets.

func Union 

func Union(this, other []uint16) []uint16

Union set operation on two sets of vbuckets, return a sorted list of vbuckets present in atleast one set.

func Vbno16to32 

func Vbno16to32(vbnos []uint16) []uint32

Vbno16to32 converts vbucket type from uint16 to uint32

func Vbno32to16 

func Vbno32to16(vbnos []uint32) []uint16

Vbno32to16 converts vbucket type from uint32 to uint16

Types

type BytesBufPool 

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

Thread safe byte buffer pool A buffer pointer received by Get() method should be put back using Put() method. This ensures that we not need to create a new buf slice with len == 0

func NewByteBufferPool 

func NewByteBufferPool(size int) *BytesBufPool

func (*BytesBufPool) Get 

func (p *BytesBufPool) Get() *[]byte

func (*BytesBufPool) Put 

func (p *BytesBufPool) Put(buf *[]byte)

type CbAuthHandler 

type CbAuthHandler struct {
	Hostport string
	Bucket   string
}

cbauth admin authentication helper Uses default cbauth env variables internally to provide auth creds

func (*CbAuthHandler) AuthenticateMemcachedConn 

func (ah *CbAuthHandler) AuthenticateMemcachedConn(host string, conn *memcached.Client) error

func (*CbAuthHandler) GetCredentials 

func (ah *CbAuthHandler) GetCredentials() (string, string)

type ClusterInfoCache 

type ClusterInfoCache struct {
	sync.Mutex
	// contains filtered or unexported fields
}

Helper object for fetching cluster information Can be used by services running on a cluster node to connect with local management service for obtaining cluster information. Info cache can be updated by using Refresh() method.

func NewClusterInfoCache 

func NewClusterInfoCache(clusterUrl string, pool string) (*ClusterInfoCache, error)

func (*ClusterInfoCache) Fetch 

func (c *ClusterInfoCache) Fetch() error

func (ClusterInfoCache) GetBucketUUID 

func (c ClusterInfoCache) GetBucketUUID(bucket string) (uuid string)

Return UUID of a given bucket.

func (ClusterInfoCache) GetCurrentNode 

func (c ClusterInfoCache) GetCurrentNode() NodeId

func (ClusterInfoCache) GetLocalHostAddress 

func (c ClusterInfoCache) GetLocalHostAddress() (string, error)

func (ClusterInfoCache) GetLocalServiceAddress 

func (c ClusterInfoCache) GetLocalServiceAddress(srvc string) (string, error)

func (ClusterInfoCache) GetLocalServiceHost 

func (c ClusterInfoCache) GetLocalServiceHost(srvc string) (string, error)

func (ClusterInfoCache) GetLocalServicePort 

func (c ClusterInfoCache) GetLocalServicePort(srvc string) (string, error)

func (ClusterInfoCache) GetNodeStatus 

func (c ClusterInfoCache) GetNodeStatus(nid NodeId) (string, error)

func (ClusterInfoCache) GetNodesByBucket 

func (c ClusterInfoCache) GetNodesByBucket(bucket string) (nids []NodeId, err error)

func (ClusterInfoCache) GetNodesByServiceType 

func (c ClusterInfoCache) GetNodesByServiceType(srvc string) (nids []NodeId)

func (ClusterInfoCache) GetServiceAddress 

func (c ClusterInfoCache) GetServiceAddress(nid NodeId, srvc string) (addr string, err error)

func (ClusterInfoCache) GetVBuckets 

func (c ClusterInfoCache) GetVBuckets(nid NodeId, bucket string) (vbs []uint32, err error)

func (ClusterInfoCache) IsNodeHealthy 

func (c ClusterInfoCache) IsNodeHealthy(nid NodeId) (bool, error)

func (*ClusterInfoCache) SetLogPrefix 

func (c *ClusterInfoCache) SetLogPrefix(p string)

func (*ClusterInfoCache) SetMaxRetries 

func (c *ClusterInfoCache) SetMaxRetries(r int)

type Config 

type Config map[string]ConfigValue

Config is a key, value map with key always being a string represents a config-parameter.

func GetSettingsConfig 

func GetSettingsConfig(cfg Config) (Config, error)

func NewConfig 

func NewConfig(data interface{}) (Config, error)

NewConfig from another Config object or from map[string]interface{} object or from []byte slice, a byte-slice of JSON string.

func (Config) Clone 

func (config Config) Clone() Config

Clone a new config object.

func (Config) FilterConfig 

func (config Config) FilterConfig(subs string) Config

func (Config) Json 

func (config Config) Json() []byte

Json will marshal config into JSON string.

func (Config) LogConfig 

func (config Config) LogConfig(prefix string)

LogConfig will check wether a configuration parameter is mutable and log that information.

func (Config) Override 

func (config Config) Override(others ...Config) Config

Override will clone `config` object and update parameters with values from `others` instance. Will skip immutable fields.

func (Config) OverrideForce 

func (config Config) OverrideForce(others ...Config) Config

OverrideForce will clone `config` object and update parameters with values from `others` instance. Will force override immutable fields as well.

func (Config) SectionConfig 

func (config Config) SectionConfig(prefix string, trim bool) Config

SectionConfig will create a new config object with parameters starting with `prefix`. If `trim` is true, then config parameter will be trimmed with the prefix string.

func (Config) Set 

func (config Config) Set(key string, cv ConfigValue) Config

Set ConfigValue for parameter. Mutates the config object.

func (Config) SetValue 

func (config Config) SetValue(key string, value interface{}) error

SetValue config parameter with value. Mutates the config object.

func (Config) Update 

func (config Config) Update(data interface{}) error

Update config object with data, can be a Config, map[string]interface{}, []byte.

type ConfigHolder 

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

Threadsafe config holder object

func (*ConfigHolder) Load 

func (h *ConfigHolder) Load() Config

func (*ConfigHolder) Store 

func (h *ConfigHolder) Store(conf Config)

type ConfigValue 

type ConfigValue struct {
	Value      interface{}
	Help       string
	DefaultVal interface{}
	Immutable  bool
}

ConfigValue for each parameter.

func (ConfigValue) Bool 

func (cv ConfigValue) Bool() bool

Bool assumes config value is a Bool and returns the same.

func (ConfigValue) Int 

func (cv ConfigValue) Int() int

Int assumes config value is an integer and returns the same.

func (ConfigValue) String 

func (cv ConfigValue) String() string

String assumes config value is a string and returns the same.

func (ConfigValue) Strings 

func (cv ConfigValue) Strings() []string

Strings assumes config value is comma separated string items.

func (ConfigValue) Uint64 

func (cv ConfigValue) Uint64() uint64

Uint64 assumes config value is 64-bit integer and returns the same.

type Consistency 

type Consistency byte

Consistency definition for index-scan queries. 

const (
	// AnyConsistency indexer would return the most current
	// data available at the moment.
	AnyConsistency Consistency = iota + 1

	// SessionConsistency indexer would query the latest timestamp
	// from each KV node. It will ensure that the scan result is at
	// least as recent as the KV timestamp. In other words, this
	// option ensures the query result is at least as recent as what
	// the user session has observed so far.
	SessionConsistency

	// QueryConsistency indexer would accept a timestamp vector,
	// and make sure to return a stable data-set that is atleast as
	// recent as the timestamp-vector.
	QueryConsistency
)

func (Consistency) String 

func (cons Consistency) String() string

type DataportKeyVersions 

type DataportKeyVersions struct {
	Bucket string
	Vbno   uint16
	Vbuuid uint64
	Kv     *KeyVersions
}

DataportKeyVersions accepted by this endpoint.

type Endpoint 

type Endpoint string //host:port

Endpoint provides an Indexer address(host:port) which is hosting a partition

type Evaluator 

type Evaluator interface {
	// Return the bucket name for which this evaluator is applicable.
	Bucket() string

	// StreamBeginData is generated for downstream.
	StreamBeginData(vbno uint16, vbuuid, seqno uint64) (data interface{})

	// Sync is generated for downstream.
	SyncData(vbno uint16, vbuuid, seqno uint64) (data interface{})

	// SnapshotData is generated for downstream.
	SnapshotData(m *mc.DcpEvent, vbno uint16, vbuuid, seqno uint64) interface{}

	// StreamEnd is generated for downstream.
	StreamEndData(vbno uint16, vbuuid, seqno uint64) (data interface{})

	// TransformRoute will transform document consumable by
	// downstream, returns data to be published to endpoints.
	TransformRoute(vbuuid uint64, m *mc.DcpEvent, data map[string]interface{}) error
}

Evaluator interface for projector, to be implemented by secondary-index or other entities.

type ExprType 

type ExprType string
const (
	JavaScript ExprType = "JavaScript"
	N1QL                = "N1QL"
)

type IndexDefn 

type IndexDefn struct {
	DefnId          IndexDefnId     `json:"defnId,omitempty"`
	Name            string          `json:"name,omitempty"`
	Using           IndexType       `json:"using,omitempty"`
	Bucket          string          `json:"bucket,omitempty"`
	BucketUUID      string          `json:"bucketUUID,omitempty"`
	IsPrimary       bool            `json:"isPrimary,omitempty"`
	SecExprs        []string        `json:"secExprs,omitempty"`
	ExprType        ExprType        `json:"exprType,omitempty"`
	PartitionScheme PartitionScheme `json:"partitionScheme,omitempty"`
	PartitionKey    string          `json:"partitionKey,omitempty"`
	WhereExpr       string          `json:"where,omitempty"`
	Deferred        bool            `json:"deferred,omitempty"`
	Nodes           []string        `json:"nodes,omitempty"`
}

IndexDefn represents the index definition as specified during CREATE INDEX

func UnmarshallIndexDefn 

func UnmarshallIndexDefn(data []byte) (*IndexDefn, error)

func (IndexDefn) String 

func (idx IndexDefn) String() string

type IndexDefnId 

type IndexDefnId uint64

func NewIndexDefnId 

func NewIndexDefnId() (IndexDefnId, error)

type IndexInst 

type IndexInst struct {
	InstId  IndexInstId
	Defn    IndexDefn
	State   IndexState
	Stream  StreamId
	Pc      PartitionContainer
	Error   string
	BuildTs []uint64
}

IndexInst is an instance of an Index(aka replica)

func (IndexInst) String 

func (idx IndexInst) String() string

type IndexInstId 

type IndexInstId uint64

type IndexInstMap 

type IndexInstMap map[IndexInstId]IndexInst

IndexInstMap is a map from IndexInstanceId to IndexInstance

func CopyIndexInstMap 

func CopyIndexInstMap(inMap IndexInstMap) IndexInstMap

func (IndexInstMap) String 

func (idx IndexInstMap) String() string

type IndexKey 

type IndexKey []byte

type IndexSnapType 

type IndexSnapType uint16

IndexSnapType represents the snapshot type created in indexer storage 

const (
	NO_SNAP IndexSnapType = iota
	DISK_SNAP
	INMEM_SNAP
)

func (IndexSnapType) String 

func (s IndexSnapType) String() string

type IndexState 

type IndexState int
const (
	//Create Index Processed
	INDEX_STATE_CREATED IndexState = iota
	// Index is stream is ready
	INDEX_STATE_READY
	//Initial Build In Progress
	INDEX_STATE_INITIAL
	//Catchup In Progress
	INDEX_STATE_CATCHUP
	//Maitenance Stream
	INDEX_STATE_ACTIVE
	//Drop Index Processed
	INDEX_STATE_DELETED
	//Error State
	INDEX_STATE_ERROR
	// Nil State (used for no-op / invalid)
	INDEX_STATE_NIL
)

func (IndexState) String 

func (s IndexState) String() string

type IndexStatistics 

type IndexStatistics interface {
	Count() (int64, error)
	MinKey() (SecondaryKey, error)
	MaxKey() (SecondaryKey, error)
	DistinctCount() (int64, error)
	Bins() ([]IndexStatistics, error)
}

IndexStatistics captures statistics for a range or a single key.

type IndexType 

type IndexType string

type IndexerId 

type IndexerId string

type KeyPartitionContainer 

type KeyPartitionContainer struct {
	PartitionMap  map[PartitionId]KeyPartitionDefn
	NumPartitions int
}

KeyPartitionContainer implements PartitionContainer interface for key based partitioning

func (*KeyPartitionContainer) AddPartition 

func (pc *KeyPartitionContainer) AddPartition(id PartitionId, p PartitionDefn)

AddPartition adds a partition to the container

func (*KeyPartitionContainer) GetAllPartitions 

func (pc *KeyPartitionContainer) GetAllPartitions() []PartitionDefn

GetAllPartitions returns all the partitions in this partitionContainer

func (*KeyPartitionContainer) GetEndpointsByPartitionId 

func (pc *KeyPartitionContainer) GetEndpointsByPartitionId(id PartitionId) []Endpoint

GetEndpointsByPartitionId returns the list of Endpoints hosting the give partitionId or nil if partitionId is not found

func (*KeyPartitionContainer) GetEndpointsByPartitionKey 

func (pc *KeyPartitionContainer) GetEndpointsByPartitionKey(key PartitionKey) []Endpoint

GetEndpointsByPartitionKey is a convenience method which calls other interface methods to first determine the partitionId from PartitionKey and then the endpoints from partitionId

func (*KeyPartitionContainer) GetNumPartitions 

func (pc *KeyPartitionContainer) GetNumPartitions() int

GetNumPartitions returns the number of partitions in this container

func (*KeyPartitionContainer) GetPartitionById 

func (pc *KeyPartitionContainer) GetPartitionById(id PartitionId) PartitionDefn

GetPartitionById returns the partition for the given partitionId or nil if partitionId is not found

func (*KeyPartitionContainer) GetPartitionIdByPartitionKey 

func (pc *KeyPartitionContainer) GetPartitionIdByPartitionKey(key PartitionKey) PartitionId

GetPartitionIdByPartitionKey returns the partitionId for the partition to which the partitionKey belongs.

func (*KeyPartitionContainer) RemovePartition 

func (pc *KeyPartitionContainer) RemovePartition(id PartitionId)

RemovePartition removes a partition from the container

func (*KeyPartitionContainer) UpdatePartition 

func (pc *KeyPartitionContainer) UpdatePartition(id PartitionId, p PartitionDefn)

UpdatePartition updates an existing partition to the container

type KeyPartitionDefn 

type KeyPartitionDefn struct {
	Id     PartitionId
	Endpts []Endpoint
}

KeyPartitionDefn defines a key based partition in terms of topology ie its Id and Indexer Endpoints hosting the partition

func (KeyPartitionDefn) Endpoints 

func (kp KeyPartitionDefn) Endpoints() []Endpoint

func (KeyPartitionDefn) GetPartitionId 

func (kp KeyPartitionDefn) GetPartitionId() PartitionId

type KeyVersions 

type KeyVersions struct {
	Seqno     uint64   // vbucket sequence number for this mutation
	Docid     []byte   // primary document id
	Uuids     []uint64 // list of unique ids, like index-ids
	Commands  []byte   // list of commands for each index
	Keys      [][]byte // list of key-versions for each index
	Oldkeys   [][]byte // previous key-versions, if available
	Partnkeys [][]byte // partition key for each key-version
}

KeyVersions for a single mutation from KV for a subset of index.

func NewKeyVersions 

func NewKeyVersions(seqno uint64, docid []byte, maxCount int) *KeyVersions

NewKeyVersions return a reference KeyVersions for a single mutation.

func (*KeyVersions) AddDeletion 

func (kv *KeyVersions) AddDeletion(uuid uint64, oldkey []byte)

AddDeletion add a new keyversion for same OpDeletion.

func (*KeyVersions) AddDropData 

func (kv *KeyVersions) AddDropData()

AddDropData add DropData command for trigger downstream catchup.

func (*KeyVersions) AddSnapshot 

func (kv *KeyVersions) AddSnapshot(typ uint32, start, end uint64)

AddSnapshot add Snapshot command for a vbucket shutdown. * type is sent via uuid field * start and end values are big-ending encoded to as key and old-key

func (*KeyVersions) AddStreamBegin 

func (kv *KeyVersions) AddStreamBegin()

AddStreamBegin add StreamBegin command for a new vbucket.

func (*KeyVersions) AddStreamEnd 

func (kv *KeyVersions) AddStreamEnd()

AddStreamEnd add StreamEnd command for a vbucket shutdown.

func (*KeyVersions) AddSync 

func (kv *KeyVersions) AddSync()

AddSync add Sync command for vbucket heartbeat.

func (*KeyVersions) AddUpsert 

func (kv *KeyVersions) AddUpsert(uuid uint64, key, oldkey []byte)

AddUpsert add a new keyversion for same OpMutation.

func (*KeyVersions) AddUpsertDeletion 

func (kv *KeyVersions) AddUpsertDeletion(uuid uint64, oldkey []byte)

AddUpsertDeletion add a keyversion command to delete old entry.

func (*KeyVersions) Equal 

func (kv *KeyVersions) Equal(other *KeyVersions) bool

Equal compares for equality of two KeyVersions object.

func (*KeyVersions) Free 

func (kv *KeyVersions) Free()

Free this object.

func (*KeyVersions) Length 

func (kv *KeyVersions) Length() int

Length number of key-versions are stored.

func (*KeyVersions) String 

func (kv *KeyVersions) String() string

type NodeId 

type NodeId int

type Notification 

type Notification struct {
	Type NotificationType
	Msg  interface{}
}

func (Notification) String 

func (n Notification) String() string

type NotificationType 

type NotificationType int
const (
	ServiceChangeNotification NotificationType = iota
	PoolChangeNotification
)

type PartitionContainer 

type PartitionContainer interface {
	AddPartition(PartitionId, PartitionDefn)
	UpdatePartition(PartitionId, PartitionDefn)
	RemovePartition(PartitionId)

	GetEndpointsByPartitionKey(PartitionKey) []Endpoint
	GetPartitionIdByPartitionKey(PartitionKey) PartitionId
	GetEndpointsByPartitionId(PartitionId) []Endpoint

	GetAllPartitions() []PartitionDefn
	GetPartitionById(PartitionId) PartitionDefn
	GetNumPartitions() int
}

PartitionContainer contains all the partitions for an index instance and provides methods to lookup topology information(i.e. endpoints) based on the PartitionKey

func NewKeyPartitionContainer 

func NewKeyPartitionContainer() PartitionContainer

NewKeyPartitionContainer initializes a new KeyPartitionContainer and returns

type PartitionDefn 

type PartitionDefn interface {
	GetPartitionId() PartitionId
	Endpoints() []Endpoint
}

PartitionDefn is a generic interface which defines a partition

type PartitionId 

type PartitionId int

type PartitionKey 

type PartitionKey []byte

type PartitionScheme 

type PartitionScheme string

type Payload 

type Payload struct {
	Payltyp byte
	Vbmap   *VbConnectionMap
	Vbs     []*VbKeyVersions // for N number of vbuckets
}

Payload either carries `vbmap` or `vbs`.

func NewStreamPayload 

func NewStreamPayload(payltyp byte, nVb int) *Payload

NewStreamPayload returns a reference to payload, `nVb` provides the maximum number of vbuckets that can be carried by a payload.

func (*Payload) AddVbKeyVersions 

func (p *Payload) AddVbKeyVersions(vb *VbKeyVersions) (err error)

AddVbKeyVersions add a VbKeyVersions as payload, one or more VbKeyVersions can be added before transport.

func (*Payload) Reset 

func (p *Payload) Reset(payltyp byte)

Reset the payload structure for next transport.

func (*Payload) SetVbmap 

func (p *Payload) SetVbmap(bucket string, vbnos []uint16, vbuuids []uint64) error

SetVbmap set vbmap as payload.

type RetryHelper 

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

Helper object to execute a function with retries and exponential backoff

func NewRetryHelper 

func NewRetryHelper(
	maxRetries int,
	interval time.Duration,
	factor int,
	call retryFunc) *RetryHelper

func (*RetryHelper) Run 

func (r *RetryHelper) Run() error

type Router 

type Router interface {
	// Bucket will return the bucket name for which this
	// router instance is applicable.
	Bucket() string

	// Endpoints return full list of endpoints <host:port>
	// that are listening for this instance.
	Endpoints() []string

	// UpsertEndpoints return a list of endpoints <host:port>
	// to which Upsert message will be published.
	//   * `key` == nil, implies missing secondary key
	//   * `partKey` == nil, implies missing partition key
	//   * m.VBucket, m.Seqno, m.Key - carry {vbno, seqno, docid}
	UpsertEndpoints(m *mc.DcpEvent, partKey, key, oldKey []byte) []string

	// UpsertDeletionEndpoints return a list of endpoints
	// <host:port> to which UpsertDeletion message will be
	// published.
	//   * `oldPartKey` and `oldKey` will be computed based on m.OldValue
	//   * `oldKey` == nil, implies old document is not available
	//   * `oldPartKey` == nil, implies old document is not available
	//   * m.VBucket, m.Seqno, m.Key - carry {vbno, seqno, docid}
	// TODO: differentiate between, missing old document and missing
	//       secondary-key
	UpsertDeletionEndpoints(m *mc.DcpEvent, oldPartKey, key, oldKey []byte) []string

	// DeletionEndpoints return a list of endpoints
	// <host:port> to which Deletion message will be published.
	//   * `oldPartKey` and `oldKey` will be computed based on m.OldValue
	//   * `oldKey` == nil, implies old document is not available
	//   * `oldPartKey` == nil, implies old document is not available
	//   * m.VBucket, m.Seqno, m.Key - carry {vbno, seqno, docid}
	// TODO: differentiate between, missing old document and missing
	//       secondary-key
	DeletionEndpoints(m *mc.DcpEvent, oldPartKey, oldKey []byte) []string
}

Router definition for each instance (aka engine), where an instance might refer to an index or similar entities.

type RouterEndpoint 

type RouterEndpoint interface {
	// Ping will check whether endpoint is active, synchronous call.
	Ping() bool

	// ResetConfig will live update configuration parameters for endpoint,
	// synchronous call.
	ResetConfig(config Config) error

	// Send will post data to endpoint client, asynchronous call.
	Send(data interface{}) error

	// GetStatistics to gather statistics information from endpoint,
	// synchronous call.
	GetStatistics() map[string]interface{}

	// Close will shutdown this endpoint and release its resources,
	// synchronous call.
	Close() error

	// WaitForExit will block until endpoint exits
	WaitForExit() error
}

RouterEndpoint abstracts downstream for feed.

type RouterEndpointFactory 

type RouterEndpointFactory func(
	topic, endpointType, raddr string, econfig Config) (RouterEndpoint, error)

RouterEndpointFactory will create a new endpoint instance for {topic, remote-address}

func (RouterEndpointFactory) MarshalJSON 

func (r RouterEndpointFactory) MarshalJSON() ([]byte, error)

MarshalJSON implements encoding/json.Marshaler{} interface

type SecondaryKey 

type SecondaryKey []interface{}

SecondaryKey is secondary-key in the shape of - [ val1, val2, ..., valN ] where value can be any golang data-type that can be serialized into JSON. simple-key shall be shaped as [ val ]

type ServiceAddressProvider 

type ServiceAddressProvider interface {
	GetLocalServiceAddress(srvc string) (string, error)
	GetLocalServicePort(srvc string) (string, error)
	GetLocalServiceHost(srvc string) (string, error)
	GetLocalHostAddress() (string, error)
}

type ServicesChangeNotifier 

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

func NewServicesChangeNotifier 

func NewServicesChangeNotifier(clusterUrl, pool string) (*ServicesChangeNotifier, error)

Initialize change notifier object for a clusterUrl

func (*ServicesChangeNotifier) Close 

func (sn *ServicesChangeNotifier) Close()

Consumer can cancel and invalidate notifier object by calling Close()

func (*ServicesChangeNotifier) Get 

func (sn *ServicesChangeNotifier) Get() (n Notification, err error)

Call Get() method to block wait and obtain next services Config

func (*ServicesChangeNotifier) GetNotifyCh 

func (sn *ServicesChangeNotifier) GetNotifyCh() chan Notification

type Statistics 

type Statistics map[string]interface{}

Statistics provide a type and method receivers for marshalling and un-marshalling statistics, as JSON, for components across the network.

func NewStatistics 

func NewStatistics(data interface{}) (stat Statistics, err error)

NewStatistics return a new instance of stat structure initialized with data.

func (Statistics) ContentType 

func (s Statistics) ContentType() string

ContentType is part of MessageMarshaller interface.

func (Statistics) Decode 

func (s Statistics) Decode(data []byte) (err error)

Decode is part of MessageMarshaller interface.

func (Statistics) Decr 

func (s Statistics) Decr(path string, vals ...int)

Decr increments stat value(s) by `vals`.

func (Statistics) Encode 

func (s Statistics) Encode() (data []byte, err error)

Encode is part of MessageMarshaller interface.

func (Statistics) Get 

func (s Statistics) Get(path string) interface{}

Get stat value

func (Statistics) Incr 

func (s Statistics) Incr(path string, vals ...int)

Incr increments stat value(s) by `vals`.

func (Statistics) Lines 

func (s Statistics) Lines() string

Lines will convert JSON to human readable list of statistics.

func (Statistics) Name 

func (s Statistics) Name() string

Name is part of MessageMarshaller interface.

func (Statistics) Set 

func (s Statistics) Set(path string, val interface{})

Set stat value

func (Statistics) ToMap 

func (s Statistics) ToMap() map[string]interface{}

ToMap converts Statistics to map.

type StreamId 

type StreamId uint16

StreamId represents the possible mutation streams 

const (
	NIL_STREAM StreamId = iota
	MAINT_STREAM
	CATCHUP_STREAM
	INIT_STREAM
	ALL_STREAMS
)

func (StreamId) String 

func (s StreamId) String() string

type TsVb 

type TsVb struct {
	Bucket string
	Vbnos  []uint16
	Seqnos []uint64
}

TsVb is logical clock for a subset of vbuckets.

type TsVbFull 

type TsVbFull struct {
	Bucket string
	Seqnos []uint64
}

TsVbFull is logical clock for full set of vbuckets.

type TsVbuuid 

type TsVbuuid struct {
	Bucket      string
	Seqnos      []uint64
	Vbuuids     []uint64
	Crc64       uint64
	Snapshots   [][2]uint64
	SnapType    IndexSnapType
	LargeSnap   bool
	SnapAligned bool
}

TsVbuuid is logical clock for full set of vbuckets along with branch value and last seen snapshot.

func NewTsVbuuid 

func NewTsVbuuid(bucket string, numVbuckets int) *TsVbuuid

NewTsVbuuid returns reference to new instance of TsVbuuid. `numVbuckets` is same as `maxVbuckets`.

func NewTsVbuuidCached 

func NewTsVbuuidCached(bucket string, numVbuckets int) *TsVbuuid

func (*TsVbuuid) AsRecent 

func (ts *TsVbuuid) AsRecent(other *TsVbuuid) bool

AsRecent will check whether timestamp `ts` is atleast as recent as timestamp `other`.

func (*TsVbuuid) AsRecentTs 

func (ts *TsVbuuid) AsRecentTs(other *TsVbuuid) bool

AsRecentTs will check whether timestamp `ts` is atleast as recent as timestamp `other`.

func (*TsVbuuid) CheckCrc64 

func (ts *TsVbuuid) CheckCrc64(other *TsVbuuid) bool

CheckVbuuids will check whether vbuuids in timestamp `ts` is same as that of `other`.

func (*TsVbuuid) CheckSnapAligned 

func (ts *TsVbuuid) CheckSnapAligned() bool

check if seqnum of all vbuckets are aligned with the snapshot end

func (*TsVbuuid) Clone 

func (ts *TsVbuuid) Clone() *TsVbuuid

Clone of TsVbuuid

func (*TsVbuuid) CompareVbuuids 

func (ts *TsVbuuid) CompareVbuuids(other *TsVbuuid) bool

CompareVbuuids will compare two timestamps for its bucket and vbuuids

func (*TsVbuuid) Copy 

func (ts *TsVbuuid) Copy() *TsVbuuid

Copy will return a clone of this timestamp.

func (*TsVbuuid) CopyFrom 

func (ts *TsVbuuid) CopyFrom(src *TsVbuuid)

func (*TsVbuuid) Diff 

func (ts *TsVbuuid) Diff(other *TsVbuuid) string

Convert the difference between two timestamps to human readable format

func (*TsVbuuid) Equal 

func (ts *TsVbuuid) Equal(other *TsVbuuid) bool

Equal returns whether `ts` and `other` compare equal.

func (*TsVbuuid) Free 

func (ts *TsVbuuid) Free()

func (*TsVbuuid) GetCrc64 

func (ts *TsVbuuid) GetCrc64() uint64

func (*TsVbuuid) GetSnapType 

func (ts *TsVbuuid) GetSnapType() IndexSnapType

Persisted returns the value of persisted flag

func (*TsVbuuid) GetVbnos 

func (ts *TsVbuuid) GetVbnos() []uint16

GetVbnos will return the list of all vbnos.

func (*TsVbuuid) HasLargeSnapshot 

func (ts *TsVbuuid) HasLargeSnapshot() bool

HasLargeSnapshot returns the value of largeSnap flag

func (*TsVbuuid) IsEpoch 

func (ts *TsVbuuid) IsEpoch() bool

func (*TsVbuuid) IsSnapAligned 

func (ts *TsVbuuid) IsSnapAligned() bool

IsSnapAligned returns the value of SnapAligned flag

func (*TsVbuuid) Len 

func (ts *TsVbuuid) Len() int

Len return number of entries in the timestamp.

func (*TsVbuuid) SetLargeSnapshot 

func (ts *TsVbuuid) SetLargeSnapshot(largeSnap bool)

SetLargeSnapshot sets the largeSnap flag

func (*TsVbuuid) SetSnapAligned 

func (ts *TsVbuuid) SetSnapAligned(snapAligned bool)

SetSnapAligned sets the SnapAligned flag

func (*TsVbuuid) SetSnapType 

func (ts *TsVbuuid) SetSnapType(typ IndexSnapType)

Persisted sets the persisted flag

func (*TsVbuuid) String 

func (ts *TsVbuuid) String() string

Convert into a human readable format

type UUID 

type UUID []byte

func NewUUID 

func NewUUID() (UUID, error)

func (UUID) Str 

func (u UUID) Str() string

func (UUID) Uint64 

func (u UUID) Uint64() uint64

type VbConnectionMap 

type VbConnectionMap struct {
	Bucket   string
	Vbuckets []uint16
	Vbuuids  []uint64
}

VbConnectionMap specifies list of vbuckets and current vbuuids for each vbucket.

func (*VbConnectionMap) Equal 

func (vbmap *VbConnectionMap) Equal(other *VbConnectionMap) bool

Equal compares to VbConnectionMap objects.

func (*VbConnectionMap) GetVbuuid 

func (vbmap *VbConnectionMap) GetVbuuid(vbno uint16) (uint64, error)

GetVbuuid returns vbuuid for specified vbucket-number from VbConnectionMap object.

type VbKeyVersions 

type VbKeyVersions struct {
	Bucket  string
	Vbucket uint16         // vbucket number
	Vbuuid  uint64         // unique id to detect branch history
	Kvs     []*KeyVersions // N number of mutations
	Uuid    string
}

VbKeyVersions carries per vbucket key-versions for one or more mutations.

func NewVbKeyVersions 

func NewVbKeyVersions(bucket string, vbno uint16, vbuuid uint64, maxMutations int) *VbKeyVersions

NewVbKeyVersions return a reference to a single vbucket payload

func (*VbKeyVersions) AddKeyVersions 

func (vb *VbKeyVersions) AddKeyVersions(kv *KeyVersions) error

AddKeyVersions will add KeyVersions for a single mutation.

func (*VbKeyVersions) Equal 

func (vb *VbKeyVersions) Equal(other *VbKeyVersions) bool

Equal compare equality of two VbKeyVersions object.

func (*VbKeyVersions) Free 

func (vb *VbKeyVersions) Free()

Free this object.

func (*VbKeyVersions) FreeKeyVersions 

func (vb *VbKeyVersions) FreeKeyVersions()

FreeKeyVersions free mutations contained by this object.

type Vbuckets 

type Vbuckets []uint16

Vbuckets is temporary data type that can be used to sort list of uint16

func (Vbuckets) Len 

func (vbuckets Vbuckets) Len() int

Len implements sort.Interface{}.

func (Vbuckets) Less 

func (vbuckets Vbuckets) Less(i, j int) bool

Less implements sort.Interface{}.

func (Vbuckets) Swap 

func (vbuckets Vbuckets) Swap(i, j int)

Swap implements sort.Interface{}

func (Vbuckets) To32 

func (vbuckets Vbuckets) To32() []uint32

To32 converts 16-bit vbucket values to 32-bit values, normally used for protobuf.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.