-
Notifications
You must be signed in to change notification settings - Fork 3.5k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Importer for 0.8.9 data via the CLI #3502
Changes from all commits
30a0ca0
05a8c10
c2f90ed
b0b229b
ea58609
76367d5
296f16d
0682c2a
b29f9a4
96aaad4
05fde32
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -7,20 +7,68 @@ import ( | |
"fmt" | ||
"io" | ||
"io/ioutil" | ||
"net" | ||
"net/http" | ||
"net/url" | ||
"strconv" | ||
"strings" | ||
"time" | ||
|
||
"github.com/influxdb/influxdb/influxql" | ||
"github.com/influxdb/influxdb/tsdb" | ||
) | ||
|
||
const ( | ||
// DefaultHost is the default host used to connect to an InfluxDB instance | ||
DefaultHost = "localhost" | ||
|
||
// DefaultPort is the default port used to connect to an InfluxDB instance | ||
DefaultPort = 8086 | ||
|
||
// DefaultTimeout is the default connection timeout used to connect to an InfluxDB instance | ||
DefaultTimeout = 0 | ||
) | ||
|
||
// Query is used to send a command to the server. Both Command and Database are required. | ||
type Query struct { | ||
Command string | ||
Database string | ||
} | ||
|
||
// ParseConnectionString will parse a string to create a valid connection URL | ||
func ParseConnectionString(path string, ssl bool) (url.URL, error) { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Could the There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. It could, but then you have to parse and validate, and if they don't give you either, you aren't sure what they wanted (you could default to http, but that might not be right). Making them explicitly choose makes it their responsibility to get right. Otherwise, if they don't provide either, it's now on us to try to make the right decision without enough information. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I was thinking that |
||
var host string | ||
var port int | ||
|
||
if strings.Contains(path, ":") { | ||
h := strings.Split(path, ":") | ||
i, e := strconv.Atoi(h[1]) | ||
if e != nil { | ||
return url.URL{}, fmt.Errorf("invalid port number %q: %s\n", path, e) | ||
} | ||
port = i | ||
if h[0] == "" { | ||
host = DefaultHost | ||
} else { | ||
host = h[0] | ||
} | ||
} else { | ||
host = path | ||
// If they didn't specify a port, always use the default port | ||
port = DefaultPort | ||
} | ||
|
||
u := url.URL{ | ||
Scheme: "http", | ||
} | ||
if ssl { | ||
u.Scheme = "https" | ||
} | ||
u.Host = net.JoinHostPort(host, strconv.Itoa(port)) | ||
|
||
return u, nil | ||
} | ||
|
||
// Config is used to specify what server to connect to. | ||
// URL: The URL of the server connecting to. | ||
// Username/Password are optional. They will be passed via basic auth if provided. | ||
|
@@ -34,6 +82,13 @@ type Config struct { | |
Timeout time.Duration | ||
} | ||
|
||
// NewConfig will create a config to be used in connecting to the client | ||
func NewConfig() *Config { | ||
return &Config{ | ||
Timeout: DefaultTimeout, | ||
} | ||
} | ||
|
||
// Client is used to make calls to the server. | ||
type Client struct { | ||
url url.URL | ||
|
@@ -51,7 +106,7 @@ const ( | |
) | ||
|
||
// NewClient will instantiate and return a connected client to issue commands to the server. | ||
func NewClient(c Config) (*Client, error) { | ||
func NewClient(c *Config) (*Client, error) { | ||
client := Client{ | ||
url: c.URL, | ||
username: c.Username, | ||
|
@@ -120,7 +175,8 @@ func (c *Client) Query(q Query) (*Response, error) { | |
// If successful, error is nil and Response is nil | ||
// If an error occurs, Response may contain additional information if populated. | ||
func (c *Client) Write(bp BatchPoints) (*Response, error) { | ||
c.url.Path = "write" | ||
u := c.url | ||
u.Path = "write" | ||
|
||
var b bytes.Buffer | ||
for _, p := range bp.Points { | ||
|
@@ -146,7 +202,7 @@ func (c *Client) Write(bp BatchPoints) (*Response, error) { | |
} | ||
} | ||
|
||
req, err := http.NewRequest("POST", c.url.String(), &b) | ||
req, err := http.NewRequest("POST", u.String(), &b) | ||
if err != nil { | ||
return nil, err | ||
} | ||
|
@@ -156,10 +212,10 @@ func (c *Client) Write(bp BatchPoints) (*Response, error) { | |
req.SetBasicAuth(c.username, c.password) | ||
} | ||
params := req.URL.Query() | ||
params.Add("db", bp.Database) | ||
params.Add("rp", bp.RetentionPolicy) | ||
params.Add("precision", bp.Precision) | ||
params.Add("consistency", bp.WriteConsistency) | ||
params.Set("db", bp.Database) | ||
params.Set("rp", bp.RetentionPolicy) | ||
params.Set("precision", bp.Precision) | ||
params.Set("consistency", bp.WriteConsistency) | ||
req.URL.RawQuery = params.Encode() | ||
|
||
resp, err := c.httpClient.Do(req) | ||
|
@@ -170,7 +226,7 @@ func (c *Client) Write(bp BatchPoints) (*Response, error) { | |
|
||
var response Response | ||
body, err := ioutil.ReadAll(resp.Body) | ||
if err != nil && err.Error() != "EOF" { | ||
if err != nil { | ||
return nil, err | ||
} | ||
|
||
|
@@ -183,6 +239,52 @@ func (c *Client) Write(bp BatchPoints) (*Response, error) { | |
return nil, nil | ||
} | ||
|
||
// WriteLineProtocol takes a string with line returns to delimit each write | ||
// If successful, error is nil and Response is nil | ||
// If an error occurs, Response may contain additional information if populated. | ||
func (c *Client) WriteLineProtocol(data, database, retentionPolicy, precision, writeConsistency string) (*Response, error) { | ||
u := c.url | ||
u.Path = "write" | ||
|
||
r := strings.NewReader(data) | ||
|
||
req, err := http.NewRequest("POST", u.String(), r) | ||
if err != nil { | ||
return nil, err | ||
} | ||
req.Header.Set("Content-Type", "") | ||
req.Header.Set("User-Agent", c.userAgent) | ||
if c.username != "" { | ||
req.SetBasicAuth(c.username, c.password) | ||
} | ||
params := req.URL.Query() | ||
params.Set("db", database) | ||
params.Set("rp", retentionPolicy) | ||
params.Set("precision", precision) | ||
params.Set("consistency", writeConsistency) | ||
req.URL.RawQuery = params.Encode() | ||
|
||
resp, err := c.httpClient.Do(req) | ||
if err != nil { | ||
return nil, err | ||
} | ||
defer resp.Body.Close() | ||
|
||
var response Response | ||
body, err := ioutil.ReadAll(resp.Body) | ||
if err != nil { | ||
return nil, err | ||
} | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
|
||
if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusOK { | ||
err := fmt.Errorf(string(body)) | ||
response.Err = err | ||
return &response, err | ||
} | ||
|
||
return nil, nil | ||
} | ||
|
||
// Ping will check to see if the server is up | ||
// Ping returns how long the request took, the version of the server it connected to, and an error if one occurred. | ||
func (c *Client) Ping() (time.Duration, string, error) { | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can you change the names to camel case? Also, spaces between the definition and the next comment line would be nice. :)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
wow, I went old school didn't I lol. Will fix!