luciano/kinesis-consumer
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

change the l4g.Critical, os.Exit combo to log.Fatal, because l4g defers output so the critical logs are not being seen

Browse source
This commit is contained in:
dan 2015-04-04 10:24:23 -07:00
parent 852cf2a7d2
commit 18163c1599
3 changed files with 10 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
pipeline.go
View file

@ -1,8 +1,8 @@
package connector package connector
import ( import (
"log"
"math" "math"
"os"
"time" "time"
"github.com/ezoic/go-kinesis" "github.com/ezoic/go-kinesis"
@ -62,8 +62,7 @@ func (p Pipeline) ProcessShard(ksis *kinesis.Kinesis, shardID string) {
shardInfo, err := ksis.GetShardIterator(args) shardInfo, err := ksis.GetShardIterator(args)
if err != nil { if err != nil {
l4g.Critical("GetShardIterator ERROR: %v", err) log.Fatalf("GetShardIterator ERROR: %v\n", err)
os.Exit(1)
} }
shardIterator := shardInfo.ShardIterator shardIterator := shardInfo.ShardIterator
@ -85,8 +84,7 @@ func (p Pipeline) ProcessShard(ksis *kinesis.Kinesis, shardID string) {
consecutiveErrorAttempts++ consecutiveErrorAttempts++
continue continue
} else { } else {
l4g.Critical("GetRecords ERROR: %v\n", err) log.Fatalf("GetRecords ERROR: %v\n", err)
os.Exit(1)
} }
} else { } else {
consecutiveErrorAttempts = 0 consecutiveErrorAttempts = 0

7
redshift_basic_emitter.go
View file

@ -4,6 +4,7 @@ import (
"bytes" "bytes"
"database/sql" "database/sql"
"fmt" "fmt"
"log"
"os" "os"
// Postgres package is used when sql.Open is called // Postgres package is used when sql.Open is called
@ -31,15 +32,13 @@ func (e RedshiftBasicEmtitter) Emit(b Buffer, t Transformer) {
db, err := sql.Open("postgres", os.Getenv("REDSHIFT_URL")) db, err := sql.Open("postgres", os.Getenv("REDSHIFT_URL"))
if err != nil { if err != nil {
l4g.Critical(err) log.Fatal(err)
os.Exit(1)
} }
_, err = db.Exec(e.copyStatement(s3File)) _, err = db.Exec(e.copyStatement(s3File))
if err != nil { if err != nil {
l4g.Critical(err) log.Fatal(err)
os.Exit(1)
} }
l4g.Debug("Redshift load completed.") l4g.Debug("Redshift load completed.")

11
redshift_manifest_emitter.go
View file

@ -5,6 +5,7 @@ import (
"database/sql" "database/sql"
"encoding/json" "encoding/json"
"fmt" "fmt"
"log"
"os" "os"
"strings" "strings"
"time" "time"
@ -35,9 +36,7 @@ func (e RedshiftManifestEmitter) Emit(b Buffer, t Transformer) {
db, err := sql.Open("postgres", os.Getenv("REDSHIFT_URL")) db, err := sql.Open("postgres", os.Getenv("REDSHIFT_URL"))
if err != nil { if err != nil {
l4g.Critical(err) log.Fatal(err)
os.Exit(1)
} }
// Aggregate file paths as strings // Aggregate file paths as strings
@ -57,8 +56,7 @@ func (e RedshiftManifestEmitter) Emit(b Buffer, t Transformer) {
_, err = db.Exec(c) _, err = db.Exec(c)
if err != nil { if err != nil {
l4g.Critical(err) log.Fatal(err)
os.Exit(1)
} }
// Insert file paths into File Names table // Insert file paths into File Names table
@ -66,8 +64,7 @@ func (e RedshiftManifestEmitter) Emit(b Buffer, t Transformer) {
_, err = db.Exec(i) _, err = db.Exec(i)
if err != nil { if err != nil {
l4g.Critical(err) log.Fatal(err)
os.Exit(1)
} }
l4g.Info("[%v] copied to Redshift", manifestFileName) l4g.Info("[%v] copied to Redshift", manifestFileName)