{"action":"created","comment":{"url":"https://api.github.com/repos/influxdb/influxdb/pulls/comments/30658510","id":30658510,"diff_hunk":"@@ -2,69 +2,252 @@ package influxdb\n \n import (\n \t\"errors\"\n+\t\"sync\"\n \t\"time\"\n+\n+\t\"github.com/influxdb/influxdb/data\"\n+\t\"github.com/influxdb/influxdb/meta\"\n+)\n+\n+const defaultWriteTimeout = 5 * time.Second\n+\n+// ConsistencyLevel represent a required replication criteria before a write can\n+// be returned as successful\n+type ConsistencyLevel int\n+\n+const (\n+\t// ConsistencyLevelAny allows for hinted hand off, potentially no write happened yet\n+\tConsistencyLevelAny ConsistencyLevel = iota\n+\n+\t// ConsistencyLevelOne requires at least one data node acknowledged a write\n+\tConsistencyLevelOne\n+\n+\t// ConsistencyLevelOne requires a quorum of data nodes to acknowledge a write\n+\tConsistencyLevelQuorum\n+\n+\t// ConsistencyLevelAll requires all data nodes to acknowledge a write\n+\tConsistencyLevelAll\n )\n \n-const defaultReadTimeout = 5 * time.Second\n+var (\n+\t// ErrTimeout is returned when a write times out\n+\tErrTimeout = errors.New(\"timeout\")\n \n-var ErrTimeout = errors.New(\"timeout\")\n+\t// ErrPartialWrite is returned when a write partially succeeds but does\n+\t// not meet the requested consistency level\n+\tErrPartialWrite = errors.New(\"partial write\")\n+\n+\t// ErrWriteFailed is returned when no writes succeeded\n+\tErrWriteFailed = errors.New(\"write failed\")\n+)\n \n // Coordinator handle queries and writes across multiple local and remote\n // data nodes.\n type Coordinator struct {\n+\tmu      sync.RWMutex\n+\tclosing chan struct{}\n+\n+\tMetaStore    meta.Store\n+\tshardWriters []ShardWriter\n+}\n+\n+// ShardMapping contiains a mapping of a shards to a points.\n+type ShardMapping struct {\n+\tPoints map[uint64][]data.Point   // The points associated with a shard ID\n+\tShards map[uint64]meta.ShardInfo // The shards that have been mapped, keyed by shard ID\n+}\n+\n+func NewCoordinator() *Coordinator {\n+\treturn &Coordinator{\n+\t\tclosing: make(chan struct{}),\n+\t}\n+}\n+\n+// NewShardMapping creates an empty ShardMapping\n+func NewShardMapping() *ShardMapping {\n+\treturn &ShardMapping{\n+\t\tPoints: map[uint64][]data.Point{},\n+\t\tShards: map[uint64]meta.ShardInfo{},\n+\t}\n+}\n+\n+// MapPoint maps a point to shard\n+func (s *ShardMapping) MapPoint(shardInfo meta.ShardInfo, p data.Point) {\n+\tpoints, ok := s.Points[shardInfo.ID]\n+\tif !ok {\n+\t\ts.Points[shardInfo.ID] = []data.Point{p}\n+\t} else {\n+\t\ts.Points[shardInfo.ID] = append(points, p)\n+\t}\n+\ts.Shards[shardInfo.ID] = shardInfo\n+}\n+\n+// ShardWriter provides the ability to write a slice of points to s given shard ID.\n+// It should return the number of times the set of points was written or an error\n+// if the write failed.\n+type ShardWriter interface {\n+\tWriteShard(shardID uint64, points []data.Point) (int, error)\n+}\n+\n+func (c *Coordinator) Open() error {\n+\tc.mu.Lock()\n+\tdefer c.mu.Unlock()\n+\tif c.closing == nil {\n+\t\tc.closing = make(chan struct{})\n+\t}\n+\treturn nil\n+}\n+\n+func (c *Coordinator) Close() error {\n+\tc.mu.Lock()\n+\tdefer c.mu.Unlock()\n+\tif c.closing != nil {\n+\t\tclose(c.closing)\n+\t\tc.closing = nil\n+\t}\n+\treturn nil\n+}\n+\n+func (c *Coordinator) AddShardWriter(s ShardWriter) {\n+\tc.mu.Lock()\n+\tdefer c.mu.Unlock()\n+\tc.shardWriters = append(c.shardWriters, s)\n+}\n+\n+// MapShards maps the points contained in wp to a ShardMapping.  If a point\n+// maps to a shard group or shard that does not currently exist, it will be\n+// created before returning the mapping.\n+func (c *Coordinator) MapShards(wp *WritePointsRequest) (*ShardMapping, error) {\n+\n+\t// holds the start time ranges for required shard groups\n+\ttimeRanges := map[time.Time]*meta.ShardGroupInfo{}\n+\n+\trp, err := c.MetaStore.RetentionPolicy(wp.Database, wp.RetentionPolicy)\n+\tif err != nil {\n+\t\treturn nil, err\n+\t}\n+\n+\tfor _, p := range wp.Points {\n+\t\ttimeRanges[p.Time.Truncate(rp.ShardGroupDuration)] = nil\n+\t}\n+\n+\t// holds all the shard groups and shards that are required for writes\n+\tfor t := range timeRanges {\n+\t\tg, err := c.MetaStore.CreateShardGroupIfNotExists(wp.Database, wp.RetentionPolicy, t)\n+\t\tif err != nil {\n+\t\t\treturn nil, err\n+\t\t}\n+\t\ttimeRanges[t] = g\n+\t}\n+\n+\tshardMapping := NewShardMapping()\n+\tfor _, p := range wp.Points {\n+\t\tg := timeRanges[p.Time.Truncate(rp.ShardGroupDuration)]\n+\t\tsid := p.SeriesID()\n+\t\tshardInfo := g.Shards[sid%uint64(len(g.Shards))]","path":"coordinator.go","position":148,"original_position":148,"commit_id":"22378523751e47842d091931ec6260ffbfe7667f","original_commit_id":"22378523751e47842d091931ec6260ffbfe7667f","user":{"login":"pauldix","id":4331,"avatar_url":"https://avatars.githubusercontent.com/u/4331?v=3","gravatar_id":"","url":"https://api.github.com/users/pauldix","html_url":"https://github.com/pauldix","followers_url":"https://api.github.com/users/pauldix/followers","following_url":"https://api.github.com/users/pauldix/following{/other_user}","gists_url":"https://api.github.com/users/pauldix/gists{/gist_id}","starred_url":"https://api.github.com/users/pauldix/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/pauldix/subscriptions","organizations_url":"https://api.github.com/users/pauldix/orgs","repos_url":"https://api.github.com/users/pauldix/repos","events_url":"https://api.github.com/users/pauldix/events{/privacy}","received_events_url":"https://api.github.com/users/pauldix/received_events","type":"User","site_admin":false},"body":"Should we extract this logic to a method `ShardGroupInfo` struct? We'll need to use it else where I think.","created_at":"2015-05-19T23:24:11Z","updated_at":"2015-05-19T23:24:11Z","html_url":"https://github.com/influxdb/influxdb/pull/2614#discussion_r30658510","pull_request_url":"https://api.github.com/repos/influxdb/influxdb/pulls/2614","_links":{"self":{"href":"https://api.github.com/repos/influxdb/influxdb/pulls/comments/30658510"},"html":{"href":"https://github.com/influxdb/influxdb/pull/2614#discussion_r30658510"},"pull_request":{"href":"https://api.github.com/repos/influxdb/influxdb/pulls/2614"}}},"pull_request":{"url":"https://api.github.com/repos/influxdb/influxdb/pulls/2614","id":35774625,"html_url":"https://github.com/influxdb/influxdb/pull/2614","diff_url":"https://github.com/influxdb/influxdb/pull/2614.diff","patch_url":"https://github.com/influxdb/influxdb/pull/2614.patch","issue_url":"https://api.github.com/repos/influxdb/influxdb/issues/2614","number":2614,"state":"open","locked":false,"title":"Initial coordinator implementation","user":{"login":"jwilder","id":219935,"avatar_url":"https://avatars.githubusercontent.com/u/219935?v=3","gravatar_id":"","url":"https://api.github.com/users/jwilder","html_url":"https://github.com/jwilder","followers_url":"https://api.github.com/users/jwilder/followers","following_url":"https://api.github.com/users/jwilder/following{/other_user}","gists_url":"https://api.github.com/users/jwilder/gists{/gist_id}","starred_url":"https://api.github.com/users/jwilder/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jwilder/subscriptions","organizations_url":"https://api.github.com/users/jwilder/orgs","repos_url":"https://api.github.com/users/jwilder/repos","events_url":"https://api.github.com/users/jwilder/events{/privacy}","received_events_url":"https://api.github.com/users/jwilder/received_events","type":"User","site_admin":false},"body":"This PR adds a `Coordinator` type that handles writes to multiple `ShardWriter`s while ensuring a consistency level for each write request.  The `Coordinator` ensures that all the required shard groups and shards have been created before a write is executed.  When a write it executed, it attempts to write to each shard in a parallel.  If the there are multiple owners for shard, then each owner is written to in parallel as well.  The write will return as soon as the first error occurs or the write level consistency is achieved.","created_at":"2015-05-19T21:03:41Z","updated_at":"2015-05-19T23:24:11Z","closed_at":null,"merged_at":null,"merge_commit_sha":"47f5a12a6ea07d9ceff77ad7573bfcae3f10a689","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/influxdb/influxdb/pulls/2614/commits","review_comments_url":"https://api.github.com/repos/influxdb/influxdb/pulls/2614/comments","review_comment_url":"https://api.github.com/repos/influxdb/influxdb/pulls/comments{/number}","comments_url":"https://api.github.com/repos/influxdb/influxdb/issues/2614/comments","statuses_url":"https://api.github.com/repos/influxdb/influxdb/statuses/22378523751e47842d091931ec6260ffbfe7667f","head":{"label":"influxdb:coordinator","ref":"coordinator","sha":"22378523751e47842d091931ec6260ffbfe7667f","user":{"login":"influxdb","id":5713248,"avatar_url":"https://avatars.githubusercontent.com/u/5713248?v=3","gravatar_id":"","url":"https://api.github.com/users/influxdb","html_url":"https://github.com/influxdb","followers_url":"https://api.github.com/users/influxdb/followers","following_url":"https://api.github.com/users/influxdb/following{/other_user}","gists_url":"https://api.github.com/users/influxdb/gists{/gist_id}","starred_url":"https://api.github.com/users/influxdb/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/influxdb/subscriptions","organizations_url":"https://api.github.com/users/influxdb/orgs","repos_url":"https://api.github.com/users/influxdb/repos","events_url":"https://api.github.com/users/influxdb/events{/privacy}","received_events_url":"https://api.github.com/users/influxdb/received_events","type":"Organization","site_admin":false},"repo":{"id":13124802,"name":"influxdb","full_name":"influxdb/influxdb","owner":{"login":"influxdb","id":5713248,"avatar_url":"https://avatars.githubusercontent.com/u/5713248?v=3","gravatar_id":"","url":"https://api.github.com/users/influxdb","html_url":"https://github.com/influxdb","followers_url":"https://api.github.com/users/influxdb/followers","following_url":"https://api.github.com/users/influxdb/following{/other_user}","gists_url":"https://api.github.com/users/influxdb/gists{/gist_id}","starred_url":"https://api.github.com/users/influxdb/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/influxdb/subscriptions","organizations_url":"https://api.github.com/users/influxdb/orgs","repos_url":"https://api.github.com/users/influxdb/repos","events_url":"https://api.github.com/users/influxdb/events{/privacy}","received_events_url":"https://api.github.com/users/influxdb/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/influxdb/influxdb","description":"Scalable datastore for metrics, events, and real-time analytics","fork":false,"url":"https://api.github.com/repos/influxdb/influxdb","forks_url":"https://api.github.com/repos/influxdb/influxdb/forks","keys_url":"https://api.github.com/repos/influxdb/influxdb/keys{/key_id}","collaborators_url":"https://api.github.com/repos/influxdb/influxdb/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/influxdb/influxdb/teams","hooks_url":"https://api.github.com/repos/influxdb/influxdb/hooks","issue_events_url":"https://api.github.com/repos/influxdb/influxdb/issues/events{/number}","events_url":"https://api.github.com/repos/influxdb/influxdb/events","assignees_url":"https://api.github.com/repos/influxdb/influxdb/assignees{/user}","branches_url":"https://api.github.com/repos/influxdb/influxdb/branches{/branch}","tags_url":"https://api.github.com/repos/influxdb/influxdb/tags","blobs_url":"https://api.github.com/repos/influxdb/influxdb/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/influxdb/influxdb/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/influxdb/influxdb/git/refs{/sha}","trees_url":"https://api.github.com/repos/influxdb/influxdb/git/trees{/sha}","statuses_url":"https://api.github.com/repos/influxdb/influxdb/statuses/{sha}","languages_url":"https://api.github.com/repos/influxdb/influxdb/languages","stargazers_url":"https://api.github.com/repos/influxdb/influxdb/stargazers","contributors_url":"https://api.github.com/repos/influxdb/influxdb/contributors","subscribers_url":"https://api.github.com/repos/influxdb/influxdb/subscribers","subscription_url":"https://api.github.com/repos/influxdb/influxdb/subscription","commits_url":"https://api.github.com/repos/influxdb/influxdb/commits{/sha}","git_commits_url":"https://api.github.com/repos/influxdb/influxdb/git/commits{/sha}","comments_url":"https://api.github.com/repos/influxdb/influxdb/comments{/number}","issue_comment_url":"https://api.github.com/repos/influxdb/influxdb/issues/comments{/number}","contents_url":"https://api.github.com/repos/influxdb/influxdb/contents/{+path}","compare_url":"https://api.github.com/repos/influxdb/influxdb/compare/{base}...{head}","merges_url":"https://api.github.com/repos/influxdb/influxdb/merges","archive_url":"https://api.github.com/repos/influxdb/influxdb/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/influxdb/influxdb/downloads","issues_url":"https://api.github.com/repos/influxdb/influxdb/issues{/number}","pulls_url":"https://api.github.com/repos/influxdb/influxdb/pulls{/number}","milestones_url":"https://api.github.com/repos/influxdb/influxdb/milestones{/number}","notifications_url":"https://api.github.com/repos/influxdb/influxdb/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/influxdb/influxdb/labels{/name}","releases_url":"https://api.github.com/repos/influxdb/influxdb/releases{/id}","created_at":"2013-09-26T14:31:10Z","updated_at":"2015-05-19T20:05:29Z","pushed_at":"2015-05-19T23:03:16Z","git_url":"git://github.com/influxdb/influxdb.git","ssh_url":"git@github.com:influxdb/influxdb.git","clone_url":"https://github.com/influxdb/influxdb.git","svn_url":"https://github.com/influxdb/influxdb","homepage":"http://influxdb.com","size":56875,"stargazers_count":5360,"watchers_count":5360,"language":"Go","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":510,"mirror_url":null,"open_issues_count":273,"forks":510,"open_issues":273,"watchers":5360,"default_branch":"master"}},"base":{"label":"influxdb:alpha1","ref":"alpha1","sha":"a682188c81c7661e8e2b63bbdd211e0b5aa82ac8","user":{"login":"influxdb","id":5713248,"avatar_url":"https://avatars.githubusercontent.com/u/5713248?v=3","gravatar_id":"","url":"https://api.github.com/users/influxdb","html_url":"https://github.com/influxdb","followers_url":"https://api.github.com/users/influxdb/followers","following_url":"https://api.github.com/users/influxdb/following{/other_user}","gists_url":"https://api.github.com/users/influxdb/gists{/gist_id}","starred_url":"https://api.github.com/users/influxdb/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/influxdb/subscriptions","organizations_url":"https://api.github.com/users/influxdb/orgs","repos_url":"https://api.github.com/users/influxdb/repos","events_url":"https://api.github.com/users/influxdb/events{/privacy}","received_events_url":"https://api.github.com/users/influxdb/received_events","type":"Organization","site_admin":false},"repo":{"id":13124802,"name":"influxdb","full_name":"influxdb/influxdb","owner":{"login":"influxdb","id":5713248,"avatar_url":"https://avatars.githubusercontent.com/u/5713248?v=3","gravatar_id":"","url":"https://api.github.com/users/influxdb","html_url":"https://github.com/influxdb","followers_url":"https://api.github.com/users/influxdb/followers","following_url":"https://api.github.com/users/influxdb/following{/other_user}","gists_url":"https://api.github.com/users/influxdb/gists{/gist_id}","starred_url":"https://api.github.com/users/influxdb/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/influxdb/subscriptions","organizations_url":"https://api.github.com/users/influxdb/orgs","repos_url":"https://api.github.com/users/influxdb/repos","events_url":"https://api.github.com/users/influxdb/events{/privacy}","received_events_url":"https://api.github.com/users/influxdb/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/influxdb/influxdb","description":"Scalable datastore for metrics, events, and real-time analytics","fork":false,"url":"https://api.github.com/repos/influxdb/influxdb","forks_url":"https://api.github.com/repos/influxdb/influxdb/forks","keys_url":"https://api.github.com/repos/influxdb/influxdb/keys{/key_id}","collaborators_url":"https://api.github.com/repos/influxdb/influxdb/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/influxdb/influxdb/teams","hooks_url":"https://api.github.com/repos/influxdb/influxdb/hooks","issue_events_url":"https://api.github.com/repos/influxdb/influxdb/issues/events{/number}","events_url":"https://api.github.com/repos/influxdb/influxdb/events","assignees_url":"https://api.github.com/repos/influxdb/influxdb/assignees{/user}","branches_url":"https://api.github.com/repos/influxdb/influxdb/branches{/branch}","tags_url":"https://api.github.com/repos/influxdb/influxdb/tags","blobs_url":"https://api.github.com/repos/influxdb/influxdb/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/influxdb/influxdb/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/influxdb/influxdb/git/refs{/sha}","trees_url":"https://api.github.com/repos/influxdb/influxdb/git/trees{/sha}","statuses_url":"https://api.github.com/repos/influxdb/influxdb/statuses/{sha}","languages_url":"https://api.github.com/repos/influxdb/influxdb/languages","stargazers_url":"https://api.github.com/repos/influxdb/influxdb/stargazers","contributors_url":"https://api.github.com/repos/influxdb/influxdb/contributors","subscribers_url":"https://api.github.com/repos/influxdb/influxdb/subscribers","subscription_url":"https://api.github.com/repos/influxdb/influxdb/subscription","commits_url":"https://api.github.com/repos/influxdb/influxdb/commits{/sha}","git_commits_url":"https://api.github.com/repos/influxdb/influxdb/git/commits{/sha}","comments_url":"https://api.github.com/repos/influxdb/influxdb/comments{/number}","issue_comment_url":"https://api.github.com/repos/influxdb/influxdb/issues/comments{/number}","contents_url":"https://api.github.com/repos/influxdb/influxdb/contents/{+path}","compare_url":"https://api.github.com/repos/influxdb/influxdb/compare/{base}...{head}","merges_url":"https://api.github.com/repos/influxdb/influxdb/merges","archive_url":"https://api.github.com/repos/influxdb/influxdb/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/influxdb/influxdb/downloads","issues_url":"https://api.github.com/repos/influxdb/influxdb/issues{/number}","pulls_url":"https://api.github.com/repos/influxdb/influxdb/pulls{/number}","milestones_url":"https://api.github.com/repos/influxdb/influxdb/milestones{/number}","notifications_url":"https://api.github.com/repos/influxdb/influxdb/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/influxdb/influxdb/labels{/name}","releases_url":"https://api.github.com/repos/influxdb/influxdb/releases{/id}","created_at":"2013-09-26T14:31:10Z","updated_at":"2015-05-19T20:05:29Z","pushed_at":"2015-05-19T23:03:16Z","git_url":"git://github.com/influxdb/influxdb.git","ssh_url":"git@github.com:influxdb/influxdb.git","clone_url":"https://github.com/influxdb/influxdb.git","svn_url":"https://github.com/influxdb/influxdb","homepage":"http://influxdb.com","size":56875,"stargazers_count":5360,"watchers_count":5360,"language":"Go","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":510,"mirror_url":null,"open_issues_count":273,"forks":510,"open_issues":273,"watchers":5360,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/influxdb/influxdb/pulls/2614"},"html":{"href":"https://github.com/influxdb/influxdb/pull/2614"},"issue":{"href":"https://api.github.com/repos/influxdb/influxdb/issues/2614"},"comments":{"href":"https://api.github.com/repos/influxdb/influxdb/issues/2614/comments"},"review_comments":{"href":"https://api.github.com/repos/influxdb/influxdb/pulls/2614/comments"},"review_comment":{"href":"https://api.github.com/repos/influxdb/influxdb/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/influxdb/influxdb/pulls/2614/commits"},"statuses":{"href":"https://api.github.com/repos/influxdb/influxdb/statuses/22378523751e47842d091931ec6260ffbfe7667f"}}}}