Skip to content
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

Инкремент №8 #8

Merged
merged 11 commits into from
Jul 29, 2024
Merged
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
added basic iter8 features
ex0rcist committed Jul 23, 2024

Verified

This commit was signed with the committer’s verified signature.
sandy081 Sandeep Somavarapu
commit 5d41387ccd5e11e04e50b9ea6c343c86dc099434
14 changes: 4 additions & 10 deletions cmd/agent/main.go
Original file line number Diff line number Diff line change
@@ -1,30 +1,24 @@
package main

import (
"context"

"github.com/ex0rcist/metflix/internal/agent"
"github.com/ex0rcist/metflix/internal/logging"
)

func main() {
ctx := logging.Setup(context.Background())
logging.Setup()

logging.LogInfo(ctx, "starting agent...")
logging.LogInfo("starting agent...")

agnt, err := agent.New()
if err != nil {
logging.LogFatal(ctx, err)
logging.LogFatal(err)
}

err = agnt.ParseFlags()
if err != nil {
logging.LogFatal(ctx, err)
logging.LogFatal(err)
}

logging.LogInfo(ctx, agnt.Config.String())

agnt.Run()

logging.LogInfo(ctx, "agent ready")
}
15 changes: 5 additions & 10 deletions cmd/server/main.go
Original file line number Diff line number Diff line change
@@ -1,32 +1,27 @@
package main

import (
"context"

"github.com/ex0rcist/metflix/internal/logging"
"github.com/ex0rcist/metflix/internal/server"
)

func main() {
ctx := logging.Setup(context.Background())
logging.Setup()

logging.LogInfo(ctx, "starting server...")
logging.LogInfo("starting server...")

srv, err := server.New()
if err != nil {
logging.LogFatal(ctx, err)
logging.LogFatal(err)
}

err = srv.ParseFlags()
if err != nil {
logging.LogFatal(ctx, err)
logging.LogFatal(err)
}

logging.LogInfo(ctx, srv.Config.String())
logging.LogInfo(ctx, "server ready") // TODO: must be after run?

err = srv.Run()
if err != nil {
logging.LogFatal(ctx, err)
logging.LogFatal(err)
}
}
3 changes: 0 additions & 3 deletions go.mod
Original file line number Diff line number Diff line change
@@ -11,7 +11,6 @@ require (
github.com/satori/go.uuid v1.2.0
github.com/spf13/pflag v1.0.5
github.com/stretchr/testify v1.9.0
github.com/tidwall/gjson v1.17.1
)

require (
@@ -20,8 +19,6 @@ require (
github.com/mattn/go-isatty v0.0.19 // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect
github.com/stretchr/objx v0.5.2 // indirect
github.com/tidwall/match v1.1.1 // indirect
github.com/tidwall/pretty v1.2.0 // indirect
golang.org/x/sys v0.21.0 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
)
6 changes: 0 additions & 6 deletions go.sum
Original file line number Diff line number Diff line change
@@ -28,12 +28,6 @@ github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY=
github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
github.com/tidwall/gjson v1.17.1 h1:wlYEnwqAHgzmhNUFfw7Xalt2JzQvsMx2Se4PcoFCT/U=
github.com/tidwall/gjson v1.17.1/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
github.com/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs=
github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
24 changes: 10 additions & 14 deletions internal/agent/agent.go
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
package agent

import (
"context"
"fmt"
"sync"
"time"
@@ -46,7 +45,7 @@ func New() (*Agent, error) {
func (a *Agent) ParseFlags() error {
address := a.Config.Address

pflag.VarP(&address, "address", "a", "address:port for HTTP API requests") // HELP: "&"" because Set() has pointer receiver?
pflag.VarP(&address, "address", "a", "address:port for HTTP API requests")

pflag.IntVarP(&a.Config.PollInterval, "poll-interval", "p", a.Config.PollInterval, "interval (s) for polling stats")
pflag.IntVarP(&a.Config.ReportInterval, "report-interval", "r", a.Config.ReportInterval, "interval (s) for polling stats")
@@ -69,6 +68,9 @@ func (a *Agent) ParseFlags() error {
}

func (a *Agent) Run() {
logging.LogInfo(a.Config.String())
logging.LogInfo("agent ready")

a.wg.Add(2)

go a.startPolling()
@@ -80,12 +82,10 @@ func (a *Agent) Run() {
func (a *Agent) startPolling() {
defer a.wg.Done()

ctx := context.Background()

for {
err := a.Stats.Poll()
if err != nil {
logging.LogError(ctx, err)
logging.LogError(err)
}

time.Sleep(intToDuration(a.Config.PollInterval))
@@ -103,9 +103,7 @@ func (a *Agent) startReporting() {
}

func (a *Agent) reportStats() {
ctx := context.Background()

logging.LogInfo(ctx, "reporting stats ... ")
logging.LogInfo("reporting stats ... ")

// agent continues polling while report is in progress, take snapshot?
snapshot := *a.Stats
@@ -154,10 +152,8 @@ func intToDuration(s int) time.Duration {
}

func (c Config) String() string {
out := "agent config: "

out += fmt.Sprintf("address=%v \t", c.Address)
out += fmt.Sprintf("poll-interval=%v \t", c.PollInterval)
out += fmt.Sprintf("report-interval=%v \t", c.ReportInterval)
return out
return fmt.Sprintf(
"agent config: address=%v; poll-interval=%v; report-interval=%v",
c.Address, c.PollInterval, c.ReportInterval,
)
}
61 changes: 48 additions & 13 deletions internal/agent/api.go
Original file line number Diff line number Diff line change
@@ -1,16 +1,18 @@
package agent

import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"time"

"github.com/ex0rcist/metflix/internal/compression"
"github.com/ex0rcist/metflix/internal/entities"
"github.com/ex0rcist/metflix/internal/logging"
"github.com/ex0rcist/metflix/internal/metrics"
"github.com/ex0rcist/metflix/internal/utils"
"github.com/rs/zerolog/log"
)

@@ -36,11 +38,10 @@ func NewAPI(address *entities.Address, httpTransport http.RoundTripper) *API {
}

func (c *API) Report(name string, metric metrics.Metric) *API {
ctx := context.Background()

// todo: another transport?
url := "http://" + c.address.String() + "/update"

var requestID = utils.GenerateRequestID()
var ctx = setupLoggerCtx(requestID)
var mex metrics.MetricExchange

// HELP: можно ли тут вместо приведения типов
@@ -51,38 +52,72 @@ func (c *API) Report(name string, metric metrics.Metric) *API {
case "gauge":
mex = metrics.NewUpdateGaugeMex(name, metric.(metrics.Gauge))
default:
log.Warn().Msg("unknown metric") // todo
logging.LogError(entities.ErrMetricReport, "unknown metric")
return c
}

body, err := json.Marshal(mex)
if err != nil {
log.Warn().Msg("unknown metric") // todo
logging.LogErrorCtx(ctx, entities.ErrMetricReport, "error during marshaling", err.Error())
return c
}

log.Info().Str("target", url).Str("payload", string(body)).Msg("sending report")
payload, err := compression.Pack(body)
if err != nil {
logging.LogErrorCtx(ctx, entities.ErrMetricReport, "error during compression", err.Error())
return c
}

req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(body))
req, err := http.NewRequest(http.MethodPost, url, payload)
if err != nil {
logging.LogError(ctx, err, "httpRequest error")
logging.LogErrorCtx(ctx, entities.ErrMetricReport, "httpRequest error", err.Error())
return c
}

req.Header.Set("Content-Type", "application/json")
req.Header.Set("Content-Encoding", "gzip")
req.Header.Set("X-Request-Id", requestID)

logRequest(ctx, url, req.Header, body)

resp, err := c.httpClient.Do(req)
if err != nil {
log.Warn().Msg("httpClient error") // todo
logging.LogErrorCtx(ctx, entities.ErrMetricReport, "error making http request", err.Error())
return c
}

defer resp.Body.Close()
respBody, err := io.ReadAll(resp.Body) // нужно прочитать ответ для keepalive?
respBody, err := io.ReadAll(resp.Body)
if err != nil {
logging.LogError(ctx, entities.ErrMetricReport, "error reading response body")
logging.LogErrorCtx(ctx, entities.ErrMetricReport, "error reading response body", err.Error())
return c
}

logResponse(ctx, resp, respBody)

if resp.StatusCode != http.StatusOK {
logging.LogError(ctx, entities.ErrMetricReport, string(respBody))
logging.LogErrorCtx(ctx, entities.ErrMetricReport, "error reporting stat", resp.Status, string(respBody))
}

return c
}

func setupLoggerCtx(requestID string) context.Context {
// empty context for now
ctx := context.Background()

// setup logger with rid attached
logger := log.Logger.With().Ctx(ctx).Str("rid", requestID).Logger()

// return context for logging
return logger.WithContext(ctx)
}

func logRequest(ctx context.Context, url string, headers http.Header, body []byte) {
logging.LogInfoCtx(ctx, "sending request to: "+url)
logging.LogDebugCtx(ctx, fmt.Sprintf("request: headers=%s; body=%s", utils.HeadersToStr(headers), string(body)))
}

func logResponse(ctx context.Context, resp *http.Response, respBody []byte) {
logging.LogDebugCtx(ctx, fmt.Sprintf("response: %v; headers=%s; body=%s", resp.Status, utils.HeadersToStr(resp.Header), respBody))
}
10 changes: 7 additions & 3 deletions internal/agent/api_test.go
Original file line number Diff line number Diff line change
@@ -6,6 +6,7 @@ import (
"net/http"
"testing"

"github.com/ex0rcist/metflix/internal/compression"
"github.com/ex0rcist/metflix/internal/entities"
"github.com/ex0rcist/metflix/internal/metrics"
"github.com/stretchr/testify/assert"
@@ -34,13 +35,16 @@ func TestApiClientReport(t *testing.T) {
assert.Equal(t, "http://localhost:8080/update", req.URL.String())
assert.Equal(t, http.MethodPost, req.Method)

expectedJSON, err := json.Marshal(metrics.NewUpdateCounterMex("test", 42))
payload, err := json.Marshal(metrics.NewUpdateCounterMex("test", 42))
require.NoError(t, err)

actualJSON, err := io.ReadAll(req.Body)
expectedPayload, err := compression.Pack(payload)
require.NoError(t, err)

assert.Equal(t, expectedJSON, actualJSON)
actualPayload, err := io.ReadAll(req.Body)
require.NoError(t, err)

assert.Equal(t, expectedPayload.Bytes(), actualPayload)

return &http.Response{
StatusCode: 200,
5 changes: 1 addition & 4 deletions internal/agent/stats.go
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
package agent

import (
"context"
"math/rand"
"time"

@@ -23,9 +22,7 @@ func NewStats() *Stats {
}

func (m *Stats) Poll() error {
ctx := context.Background()

logging.LogInfo(ctx, "polling stats ... ")
logging.LogDebug("polling stats ... ")

m.PollCount++
m.RandomValue = metrics.Gauge(m.generator.Float64())
26 changes: 26 additions & 0 deletions internal/compression/compression.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
package compression

import (
"bytes"
"compress/gzip"
"fmt"
)

func Pack(data []byte) (*bytes.Buffer, error) {
bb := new(bytes.Buffer)

encoder, err := gzip.NewWriterLevel(bb, gzip.BestSpeed)
if err != nil {
return nil, fmt.Errorf("failed init compress writer: %v", err)
}

if _, err = encoder.Write(data); err != nil {
return nil, fmt.Errorf("failed write data to compress temporary buffer: %v", err)
}

if err = encoder.Close(); err != nil {
return nil, fmt.Errorf("failed compress data: %v", err)
}

return bb, nil
}
Loading