{"action":"created","comment":{"url":"https://api.github.com/repos/remind101/logspout-kinesis/pulls/comments/39320665","id":39320665,"diff_hunk":"@@ -11,142 +10,103 @@ import (\n \t\"github.com/pborman/uuid\"\n )\n \n-// PutRecordsLimit is the maximum number of records allowed for a PutRecords request.\n-var PutRecordsLimit = 500\n+var ErrRecordTooBig = errors.New(\"data byte size is over the limit\")\n \n-// RecordSizeLimit is the maximum allowed size per record.\n-var RecordSizeLimit int = 1 * 1024 * 1024 // 1MB\n+type limits struct {\n+\t// PutRecordsLimit is the maximum number of records allowed for a PutRecords request.\n+\tputRecords int\n \n-// PutRecordsSizeLimit is the maximum allowed size per PutRecords request.\n-var PutRecordsSizeLimit int = 5 * 1024 * 1024 // 5MB\n+\t// PutRecordsSizeLimit is the maximum allowed size per PutRecords request.\n+\tputRecordsSize int\n \n-type recordBuffer struct {\n-\tclient   *kinesis.Kinesis\n-\tpKeyTmpl *template.Template\n-\tinput    *kinesis.PutRecordsInput\n-\tcount    int\n-\tbyteSize int\n-\tmutex    sync.Mutex\n+\t// RecordSizeLimit is the maximum allowed size per record.\n+\trecordSize int\n }\n \n-func newRecordBuffer(client *kinesis.Kinesis, streamName string) (*recordBuffer, error) {\n-\tpKeyTmpl, err := compileTmpl(\"KINESIS_PARTITION_KEY_TEMPLATE\")\n-\tif err != nil {\n-\t\treturn nil, err\n-\t}\n-\n-\tinput := &kinesis.PutRecordsInput{\n-\t\tStreamName: aws.String(streamName),\n-\t\tRecords:    make([]*kinesis.PutRecordsRequestEntry, 0),\n-\t}\n-\n-\treturn &recordBuffer{\n-\t\tclient:   client,\n-\t\tpKeyTmpl: pKeyTmpl,\n-\t\tinput:    input,\n-\t}, nil\n-}\n+const (\n+\tPutRecordsLimit     int = 500\n+\tPutRecordsSizeLimit int = 5 * 1024 * 1024 // 5MB\n+\tRecordSizeLimit     int = 1 * 1024 * 1024 // 1MB\n+)\n \n-type recordSizeLimitError struct {\n-\tcaller string\n-\tlength int\n+type buffer struct {\n+\tcount    int\n+\tbyteSize int\n+\tpKeyTmpl *template.Template\n+\tinput    kinesis.PutRecordsInput\n+\tlimits   *limits\n }\n \n-func (e *recordSizeLimitError) Error() string {\n-\treturn fmt.Sprintf(\"%s: log data byte size (%d) is over the limit.\", e.caller, e.length)\n+func newBuffer(tmpl *template.Template, sn string) *buffer {\n+\treturn &buffer{\n+\t\tpKeyTmpl: tmpl,\n+\t\tinput: kinesis.PutRecordsInput{\n+\t\t\tStreamName: aws.String(sn),\n+\t\t\tRecords:    make([]*kinesis.PutRecordsRequestEntry, 0),\n+\t\t},\n+\t\tlimits: &limits{\n+\t\t\tputRecords:     PutRecordsLimit,\n+\t\t\tputRecordsSize: PutRecordsSizeLimit,\n+\t\t\trecordSize:     RecordSizeLimit,\n+\t\t},\n+\t}\n }\n \n-// Add fills the buffer with new data, or flushes it if one of the limit\n-// has hit.\n-func (r *recordBuffer) Add(m *router.Message) error {\n-\tdata := m.Data\n-\tdataLen := len(data)\n+func (b *buffer) add(m *router.Message) error {","path":"buffer.go","position":96,"original_position":96,"commit_id":"5e006f5f9f09d75d4ffe0143de4162967e17a2ea","original_commit_id":"5e006f5f9f09d75d4ffe0143de4162967e17a2ea","user":{"login":"benguillet","id":766416,"avatar_url":"https://avatars.githubusercontent.com/u/766416?v=3","gravatar_id":"","url":"https://api.github.com/users/benguillet","html_url":"https://github.com/benguillet","followers_url":"https://api.github.com/users/benguillet/followers","following_url":"https://api.github.com/users/benguillet/following{/other_user}","gists_url":"https://api.github.com/users/benguillet/gists{/gist_id}","starred_url":"https://api.github.com/users/benguillet/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/benguillet/subscriptions","organizations_url":"https://api.github.com/users/benguillet/orgs","repos_url":"https://api.github.com/users/benguillet/repos","events_url":"https://api.github.com/users/benguillet/events{/privacy}","received_events_url":"https://api.github.com/users/benguillet/received_events","type":"User","site_admin":false},"body":"No, it's only called by the bufferMessages goroutine, and then we sent a copy of the buffer to `w.buffers`. I wonder if we could lose a message if the add() hasn't finished by the time a `reset()` is triggered by the ticker though...","created_at":"2015-09-11T21:38:58Z","updated_at":"2015-09-11T21:38:58Z","html_url":"https://github.com/remind101/logspout-kinesis/pull/16#discussion_r39320665","pull_request_url":"https://api.github.com/repos/remind101/logspout-kinesis/pulls/16","_links":{"self":{"href":"https://api.github.com/repos/remind101/logspout-kinesis/pulls/comments/39320665"},"html":{"href":"https://github.com/remind101/logspout-kinesis/pull/16#discussion_r39320665"},"pull_request":{"href":"https://api.github.com/repos/remind101/logspout-kinesis/pulls/16"}}},"pull_request":{"url":"https://api.github.com/repos/remind101/logspout-kinesis/pulls/16","id":44064943,"html_url":"https://github.com/remind101/logspout-kinesis/pull/16","diff_url":"https://github.com/remind101/logspout-kinesis/pull/16.diff","patch_url":"https://github.com/remind101/logspout-kinesis/pull/16.patch","issue_url":"https://api.github.com/repos/remind101/logspout-kinesis/issues/16","number":16,"state":"open","locked":false,"title":"Fix pump timeout","user":{"login":"benguillet","id":766416,"avatar_url":"https://avatars.githubusercontent.com/u/766416?v=3","gravatar_id":"","url":"https://api.github.com/users/benguillet","html_url":"https://github.com/benguillet","followers_url":"https://api.github.com/users/benguillet/followers","following_url":"https://api.github.com/users/benguillet/following{/other_user}","gists_url":"https://api.github.com/users/benguillet/gists{/gist_id}","starred_url":"https://api.github.com/users/benguillet/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/benguillet/subscriptions","organizations_url":"https://api.github.com/users/benguillet/orgs","repos_url":"https://api.github.com/users/benguillet/repos","events_url":"https://api.github.com/users/benguillet/events{/privacy}","received_events_url":"https://api.github.com/users/benguillet/received_events","type":"User","site_admin":false},"body":"The old architecture was blocking in some conditions and thus triggering a pump timeout in logspout, which results in logspout deleting the logstream channel and completely stoppping to send logs to the adapter.\r\n\r\nThis is a refactor using channels, select and better goroutines, so that it never blocks. If the flush takes too long and the flusher start to accumulate delay (more than 10 buffers in the queue), it starts dropping buffers (with a warning). This refactor also adds tests.","created_at":"2015-09-03T02:55:14Z","updated_at":"2015-09-11T21:38:58Z","closed_at":null,"merged_at":null,"merge_commit_sha":"f152c35cbd16800c05fce71e96fd21a5cac35859","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/remind101/logspout-kinesis/pulls/16/commits","review_comments_url":"https://api.github.com/repos/remind101/logspout-kinesis/pulls/16/comments","review_comment_url":"https://api.github.com/repos/remind101/logspout-kinesis/pulls/comments{/number}","comments_url":"https://api.github.com/repos/remind101/logspout-kinesis/issues/16/comments","statuses_url":"https://api.github.com/repos/remind101/logspout-kinesis/statuses/5e006f5f9f09d75d4ffe0143de4162967e17a2ea","head":{"label":"remind101:fix-pump-timeout","ref":"fix-pump-timeout","sha":"5e006f5f9f09d75d4ffe0143de4162967e17a2ea","user":{"login":"remind101","id":376343,"avatar_url":"https://avatars.githubusercontent.com/u/376343?v=3","gravatar_id":"","url":"https://api.github.com/users/remind101","html_url":"https://github.com/remind101","followers_url":"https://api.github.com/users/remind101/followers","following_url":"https://api.github.com/users/remind101/following{/other_user}","gists_url":"https://api.github.com/users/remind101/gists{/gist_id}","starred_url":"https://api.github.com/users/remind101/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/remind101/subscriptions","organizations_url":"https://api.github.com/users/remind101/orgs","repos_url":"https://api.github.com/users/remind101/repos","events_url":"https://api.github.com/users/remind101/events{/privacy}","received_events_url":"https://api.github.com/users/remind101/received_events","type":"Organization","site_admin":false},"repo":{"id":39969401,"name":"logspout-kinesis","full_name":"remind101/logspout-kinesis","owner":{"login":"remind101","id":376343,"avatar_url":"https://avatars.githubusercontent.com/u/376343?v=3","gravatar_id":"","url":"https://api.github.com/users/remind101","html_url":"https://github.com/remind101","followers_url":"https://api.github.com/users/remind101/followers","following_url":"https://api.github.com/users/remind101/following{/other_user}","gists_url":"https://api.github.com/users/remind101/gists{/gist_id}","starred_url":"https://api.github.com/users/remind101/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/remind101/subscriptions","organizations_url":"https://api.github.com/users/remind101/orgs","repos_url":"https://api.github.com/users/remind101/repos","events_url":"https://api.github.com/users/remind101/events{/privacy}","received_events_url":"https://api.github.com/users/remind101/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/remind101/logspout-kinesis","description":"A Logspout adapter for writing logs to Amazon Kinesis","fork":false,"url":"https://api.github.com/repos/remind101/logspout-kinesis","forks_url":"https://api.github.com/repos/remind101/logspout-kinesis/forks","keys_url":"https://api.github.com/repos/remind101/logspout-kinesis/keys{/key_id}","collaborators_url":"https://api.github.com/repos/remind101/logspout-kinesis/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/remind101/logspout-kinesis/teams","hooks_url":"https://api.github.com/repos/remind101/logspout-kinesis/hooks","issue_events_url":"https://api.github.com/repos/remind101/logspout-kinesis/issues/events{/number}","events_url":"https://api.github.com/repos/remind101/logspout-kinesis/events","assignees_url":"https://api.github.com/repos/remind101/logspout-kinesis/assignees{/user}","branches_url":"https://api.github.com/repos/remind101/logspout-kinesis/branches{/branch}","tags_url":"https://api.github.com/repos/remind101/logspout-kinesis/tags","blobs_url":"https://api.github.com/repos/remind101/logspout-kinesis/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/remind101/logspout-kinesis/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/remind101/logspout-kinesis/git/refs{/sha}","trees_url":"https://api.github.com/repos/remind101/logspout-kinesis/git/trees{/sha}","statuses_url":"https://api.github.com/repos/remind101/logspout-kinesis/statuses/{sha}","languages_url":"https://api.github.com/repos/remind101/logspout-kinesis/languages","stargazers_url":"https://api.github.com/repos/remind101/logspout-kinesis/stargazers","contributors_url":"https://api.github.com/repos/remind101/logspout-kinesis/contributors","subscribers_url":"https://api.github.com/repos/remind101/logspout-kinesis/subscribers","subscription_url":"https://api.github.com/repos/remind101/logspout-kinesis/subscription","commits_url":"https://api.github.com/repos/remind101/logspout-kinesis/commits{/sha}","git_commits_url":"https://api.github.com/repos/remind101/logspout-kinesis/git/commits{/sha}","comments_url":"https://api.github.com/repos/remind101/logspout-kinesis/comments{/number}","issue_comment_url":"https://api.github.com/repos/remind101/logspout-kinesis/issues/comments{/number}","contents_url":"https://api.github.com/repos/remind101/logspout-kinesis/contents/{+path}","compare_url":"https://api.github.com/repos/remind101/logspout-kinesis/compare/{base}...{head}","merges_url":"https://api.github.com/repos/remind101/logspout-kinesis/merges","archive_url":"https://api.github.com/repos/remind101/logspout-kinesis/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/remind101/logspout-kinesis/downloads","issues_url":"https://api.github.com/repos/remind101/logspout-kinesis/issues{/number}","pulls_url":"https://api.github.com/repos/remind101/logspout-kinesis/pulls{/number}","milestones_url":"https://api.github.com/repos/remind101/logspout-kinesis/milestones{/number}","notifications_url":"https://api.github.com/repos/remind101/logspout-kinesis/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/remind101/logspout-kinesis/labels{/name}","releases_url":"https://api.github.com/repos/remind101/logspout-kinesis/releases{/id}","created_at":"2015-07-30T19:49:29Z","updated_at":"2015-09-07T21:39:58Z","pushed_at":"2015-09-11T17:57:19Z","git_url":"git://github.com/remind101/logspout-kinesis.git","ssh_url":"git@github.com:remind101/logspout-kinesis.git","clone_url":"https://github.com/remind101/logspout-kinesis.git","svn_url":"https://github.com/remind101/logspout-kinesis","homepage":null,"size":676,"stargazers_count":2,"watchers_count":2,"language":"Go","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":4,"forks":0,"open_issues":4,"watchers":2,"default_branch":"master"}},"base":{"label":"remind101:master","ref":"master","sha":"6fc8366175ae6ee46af8c7a9438a3f45ce5699d7","user":{"login":"remind101","id":376343,"avatar_url":"https://avatars.githubusercontent.com/u/376343?v=3","gravatar_id":"","url":"https://api.github.com/users/remind101","html_url":"https://github.com/remind101","followers_url":"https://api.github.com/users/remind101/followers","following_url":"https://api.github.com/users/remind101/following{/other_user}","gists_url":"https://api.github.com/users/remind101/gists{/gist_id}","starred_url":"https://api.github.com/users/remind101/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/remind101/subscriptions","organizations_url":"https://api.github.com/users/remind101/orgs","repos_url":"https://api.github.com/users/remind101/repos","events_url":"https://api.github.com/users/remind101/events{/privacy}","received_events_url":"https://api.github.com/users/remind101/received_events","type":"Organization","site_admin":false},"repo":{"id":39969401,"name":"logspout-kinesis","full_name":"remind101/logspout-kinesis","owner":{"login":"remind101","id":376343,"avatar_url":"https://avatars.githubusercontent.com/u/376343?v=3","gravatar_id":"","url":"https://api.github.com/users/remind101","html_url":"https://github.com/remind101","followers_url":"https://api.github.com/users/remind101/followers","following_url":"https://api.github.com/users/remind101/following{/other_user}","gists_url":"https://api.github.com/users/remind101/gists{/gist_id}","starred_url":"https://api.github.com/users/remind101/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/remind101/subscriptions","organizations_url":"https://api.github.com/users/remind101/orgs","repos_url":"https://api.github.com/users/remind101/repos","events_url":"https://api.github.com/users/remind101/events{/privacy}","received_events_url":"https://api.github.com/users/remind101/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/remind101/logspout-kinesis","description":"A Logspout adapter for writing logs to Amazon Kinesis","fork":false,"url":"https://api.github.com/repos/remind101/logspout-kinesis","forks_url":"https://api.github.com/repos/remind101/logspout-kinesis/forks","keys_url":"https://api.github.com/repos/remind101/logspout-kinesis/keys{/key_id}","collaborators_url":"https://api.github.com/repos/remind101/logspout-kinesis/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/remind101/logspout-kinesis/teams","hooks_url":"https://api.github.com/repos/remind101/logspout-kinesis/hooks","issue_events_url":"https://api.github.com/repos/remind101/logspout-kinesis/issues/events{/number}","events_url":"https://api.github.com/repos/remind101/logspout-kinesis/events","assignees_url":"https://api.github.com/repos/remind101/logspout-kinesis/assignees{/user}","branches_url":"https://api.github.com/repos/remind101/logspout-kinesis/branches{/branch}","tags_url":"https://api.github.com/repos/remind101/logspout-kinesis/tags","blobs_url":"https://api.github.com/repos/remind101/logspout-kinesis/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/remind101/logspout-kinesis/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/remind101/logspout-kinesis/git/refs{/sha}","trees_url":"https://api.github.com/repos/remind101/logspout-kinesis/git/trees{/sha}","statuses_url":"https://api.github.com/repos/remind101/logspout-kinesis/statuses/{sha}","languages_url":"https://api.github.com/repos/remind101/logspout-kinesis/languages","stargazers_url":"https://api.github.com/repos/remind101/logspout-kinesis/stargazers","contributors_url":"https://api.github.com/repos/remind101/logspout-kinesis/contributors","subscribers_url":"https://api.github.com/repos/remind101/logspout-kinesis/subscribers","subscription_url":"https://api.github.com/repos/remind101/logspout-kinesis/subscription","commits_url":"https://api.github.com/repos/remind101/logspout-kinesis/commits{/sha}","git_commits_url":"https://api.github.com/repos/remind101/logspout-kinesis/git/commits{/sha}","comments_url":"https://api.github.com/repos/remind101/logspout-kinesis/comments{/number}","issue_comment_url":"https://api.github.com/repos/remind101/logspout-kinesis/issues/comments{/number}","contents_url":"https://api.github.com/repos/remind101/logspout-kinesis/contents/{+path}","compare_url":"https://api.github.com/repos/remind101/logspout-kinesis/compare/{base}...{head}","merges_url":"https://api.github.com/repos/remind101/logspout-kinesis/merges","archive_url":"https://api.github.com/repos/remind101/logspout-kinesis/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/remind101/logspout-kinesis/downloads","issues_url":"https://api.github.com/repos/remind101/logspout-kinesis/issues{/number}","pulls_url":"https://api.github.com/repos/remind101/logspout-kinesis/pulls{/number}","milestones_url":"https://api.github.com/repos/remind101/logspout-kinesis/milestones{/number}","notifications_url":"https://api.github.com/repos/remind101/logspout-kinesis/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/remind101/logspout-kinesis/labels{/name}","releases_url":"https://api.github.com/repos/remind101/logspout-kinesis/releases{/id}","created_at":"2015-07-30T19:49:29Z","updated_at":"2015-09-07T21:39:58Z","pushed_at":"2015-09-11T17:57:19Z","git_url":"git://github.com/remind101/logspout-kinesis.git","ssh_url":"git@github.com:remind101/logspout-kinesis.git","clone_url":"https://github.com/remind101/logspout-kinesis.git","svn_url":"https://github.com/remind101/logspout-kinesis","homepage":null,"size":676,"stargazers_count":2,"watchers_count":2,"language":"Go","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":4,"forks":0,"open_issues":4,"watchers":2,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/remind101/logspout-kinesis/pulls/16"},"html":{"href":"https://github.com/remind101/logspout-kinesis/pull/16"},"issue":{"href":"https://api.github.com/repos/remind101/logspout-kinesis/issues/16"},"comments":{"href":"https://api.github.com/repos/remind101/logspout-kinesis/issues/16/comments"},"review_comments":{"href":"https://api.github.com/repos/remind101/logspout-kinesis/pulls/16/comments"},"review_comment":{"href":"https://api.github.com/repos/remind101/logspout-kinesis/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/remind101/logspout-kinesis/pulls/16/commits"},"statuses":{"href":"https://api.github.com/repos/remind101/logspout-kinesis/statuses/5e006f5f9f09d75d4ffe0143de4162967e17a2ea"}}}}