fix pipeline (#365)
Co-authored-by: Aleksandr Tolstikhin <atolstikhin@mtsbank.ru> Reviewed-on: #365 Co-authored-by: Vasiliy Tolstov <v.tolstov@unistack.org> Co-committed-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
@@ -66,8 +66,13 @@ func (c *dnsConn) RemoteAddr() net.Addr {
|
||||
}
|
||||
|
||||
func (c *dnsConn) SetDeadline(t time.Time) error {
|
||||
c.SetReadDeadline(t)
|
||||
c.SetWriteDeadline(t)
|
||||
var err error
|
||||
if err = c.SetReadDeadline(t); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = c.SetWriteDeadline(t); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -16,7 +16,6 @@ type Ticker struct {
|
||||
C chan time.Time
|
||||
min int64
|
||||
max int64
|
||||
exp int64
|
||||
exit bool
|
||||
rng rand.Rand
|
||||
}
|
||||
|
||||
@@ -91,7 +91,7 @@ func Merge(dst interface{}, mp map[string]interface{}, opts ...Option) error {
|
||||
}
|
||||
|
||||
if mapper, ok := dst.(map[string]interface{}); ok {
|
||||
dst = mergeMap(mapper, mp, 0)
|
||||
mergeMap(mapper, mp, 0)
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -1,9 +1,38 @@
|
||||
package reflect
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestMergeMapStringInterface(t *testing.T) {
|
||||
var dst interface{} //nolint:gosimple
|
||||
dst = map[string]interface{}{
|
||||
"xx": 11,
|
||||
}
|
||||
|
||||
src := map[string]interface{}{
|
||||
"zz": "aa",
|
||||
}
|
||||
|
||||
if err := Merge(dst, src); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
mp, ok := dst.(map[string]interface{})
|
||||
if !ok || mp == nil {
|
||||
t.Fatalf("xxx %#+v\n", dst)
|
||||
}
|
||||
|
||||
if fmt.Sprintf("%v", mp["xx"]) != "11" {
|
||||
t.Fatalf("xxx zzzz %#+v", mp)
|
||||
}
|
||||
|
||||
if fmt.Sprintf("%v", mp["zz"]) != "aa" {
|
||||
t.Fatalf("xxx zzzz %#+v", mp)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeMap(t *testing.T) {
|
||||
src := map[string]interface{}{
|
||||
"skey1": "sval1",
|
||||
|
||||
@@ -56,7 +56,7 @@ type DigitalOceanMetadata struct {
|
||||
func (stfs *DigitalOceanMetadata) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.URL.Path {
|
||||
case "/metadata/v1.json":
|
||||
json.NewEncoder(w).Encode(stfs.Metadata.V1)
|
||||
_ = json.NewEncoder(w).Encode(stfs.Metadata.V1)
|
||||
default:
|
||||
fs := FileServer(stfs, "json", time.Now())
|
||||
idx := strings.Index(r.URL.Path[1:], "/")
|
||||
|
||||
@@ -12,7 +12,7 @@ type EC2Metadata struct {
|
||||
InstanceType string `json:"instance-type"`
|
||||
LocalHostname string `json:"local-hostname"`
|
||||
LocalIPv4 string `json:"local-ipv4"`
|
||||
kernelID int `json:"kernel-id"`
|
||||
KernelID int `json:"kernel-id"`
|
||||
Placement string `json:"placement"`
|
||||
AvailabilityZone string `json:"availability-zone"`
|
||||
ProductCodes string `json:"product-codes"`
|
||||
|
||||
@@ -27,7 +27,7 @@ func (fs *fs) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
f, err := fs.Open(r.URL.Path)
|
||||
if err != nil {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
w.Write([]byte(err.Error()))
|
||||
_, _ = w.Write([]byte(err.Error()))
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/octet-stream")
|
||||
@@ -67,9 +67,9 @@ func (fi *fileInfo) Name() string {
|
||||
|
||||
func (fi *fileInfo) Mode() os.FileMode {
|
||||
if strings.HasSuffix(fi.name, "/") {
|
||||
return os.FileMode(0755) | os.ModeDir
|
||||
return os.FileMode(0o755) | os.ModeDir
|
||||
}
|
||||
return os.FileMode(0644)
|
||||
return os.FileMode(0o644)
|
||||
}
|
||||
|
||||
func (fi *fileInfo) IsDir() bool {
|
||||
@@ -112,15 +112,14 @@ func (f *file) Readdir(count int) ([]os.FileInfo, error) {
|
||||
func (f *file) Seek(offset int64, whence int) (int64, error) {
|
||||
// log.Printf("seek %d %d %s\n", offset, whence, f.name)
|
||||
switch whence {
|
||||
case os.SEEK_SET:
|
||||
case io.SeekStart:
|
||||
f.offset = offset
|
||||
case os.SEEK_CUR:
|
||||
case io.SeekCurrent:
|
||||
f.offset += offset
|
||||
case os.SEEK_END:
|
||||
case io.SeekEnd:
|
||||
f.offset = int64(len(f.data)) + offset
|
||||
}
|
||||
return f.offset, nil
|
||||
|
||||
}
|
||||
|
||||
func (f *file) Stat() (os.FileInfo, error) {
|
||||
@@ -222,6 +221,7 @@ func getValue(name string, iface interface{}, tag string) ([]byte, error) {
|
||||
return nil, fmt.Errorf("failed to find %s in interface %T", name, iface)
|
||||
}
|
||||
|
||||
/*
|
||||
func hasValidType(obj interface{}, types []reflect.Kind) bool {
|
||||
for _, t := range types {
|
||||
if reflect.TypeOf(obj).Kind() == t {
|
||||
@@ -231,6 +231,7 @@ func hasValidType(obj interface{}, types []reflect.Kind) bool {
|
||||
|
||||
return false
|
||||
}
|
||||
*/
|
||||
|
||||
func reflectValue(obj interface{}) reflect.Value {
|
||||
var val reflect.Value
|
||||
|
||||
@@ -2,7 +2,7 @@ package structfs
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"io/ioutil"
|
||||
"io"
|
||||
"net/http"
|
||||
"reflect"
|
||||
"testing"
|
||||
@@ -61,7 +61,7 @@ var doOrig = []byte(`{
|
||||
}
|
||||
`)
|
||||
|
||||
func server(t *testing.T) {
|
||||
func server(t *testing.T, ch chan error) {
|
||||
stfs := DigitalOceanMetadata{}
|
||||
err := json.Unmarshal(doOrig, &stfs.Metadata.V1)
|
||||
if err != nil {
|
||||
@@ -71,7 +71,7 @@ func server(t *testing.T) {
|
||||
http.Handle("/metadata/v1/", FileServer(&stfs, "json", time.Now()))
|
||||
http.Handle("/metadata/v1.json", &stfs)
|
||||
go func() {
|
||||
t.Fatal(http.ListenAndServe("127.0.0.1:8080", nil))
|
||||
ch <- http.ListenAndServe("127.0.0.1:8080", nil)
|
||||
}()
|
||||
time.Sleep(2 * time.Second)
|
||||
}
|
||||
@@ -82,13 +82,14 @@ func get(path string) ([]byte, error) {
|
||||
return nil, err
|
||||
}
|
||||
defer res.Body.Close()
|
||||
return ioutil.ReadAll(res.Body)
|
||||
return io.ReadAll(res.Body)
|
||||
}
|
||||
|
||||
func TestAll(t *testing.T) {
|
||||
server(t)
|
||||
ch := make(chan error)
|
||||
server(t, ch)
|
||||
|
||||
var tests = []struct {
|
||||
tests := []struct {
|
||||
in string
|
||||
out string
|
||||
}{
|
||||
@@ -100,34 +101,44 @@ func TestAll(t *testing.T) {
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
buf, err := get(tt.in)
|
||||
select {
|
||||
case err := <-ch:
|
||||
t.Fatal(err)
|
||||
default:
|
||||
buf, err := get(tt.in)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if string(buf) != tt.out {
|
||||
t.Errorf("req %s output %s not match requested %s", tt.in, string(buf), tt.out)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
select {
|
||||
case err := <-ch:
|
||||
t.Fatal(err)
|
||||
default:
|
||||
doTest, err := get("http://127.0.0.1:8080/metadata/v1.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if string(buf) != tt.out {
|
||||
t.Errorf("req %s output %s not match requested %s", tt.in, string(buf), tt.out)
|
||||
|
||||
oSt := DigitalOceanMetadata{}
|
||||
err = json.Unmarshal(doOrig, &oSt.Metadata.V1)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
nSt := DigitalOceanMetadata{}
|
||||
|
||||
err = json.Unmarshal(doTest, &nSt.Metadata.V1)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if !reflect.DeepEqual(oSt, nSt) {
|
||||
t.Fatalf("%v not match %v", oSt, nSt)
|
||||
}
|
||||
}
|
||||
|
||||
doTest, err := get("http://127.0.0.1:8080/metadata/v1.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
oSt := DigitalOceanMetadata{}
|
||||
err = json.Unmarshal(doOrig, &oSt.Metadata.V1)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
nSt := DigitalOceanMetadata{}
|
||||
|
||||
err = json.Unmarshal(doTest, &nSt.Metadata.V1)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if !reflect.DeepEqual(oSt, nSt) {
|
||||
t.Fatalf("%v not match %v", oSt, nSt)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -14,7 +14,7 @@ type Duration int64
|
||||
|
||||
func ParseDuration(s string) (time.Duration, error) {
|
||||
if s == "" {
|
||||
return 0, fmt.Errorf(`time: invalid duration "` + s + `"`)
|
||||
return 0, errors.New(`time: invalid duration "` + s + `"`)
|
||||
}
|
||||
|
||||
var p int
|
||||
|
||||
@@ -39,19 +39,16 @@ func newStatsMeter() {
|
||||
ticker := time.NewTicker(meter.DefaultMeterStatsInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
poolsMu.Lock()
|
||||
for _, st := range pools {
|
||||
stats := st.Stats()
|
||||
meter.DefaultMeter.Counter(semconv.PoolGetTotal, "capacity", strconv.Itoa(st.Cap())).Set(stats.Get)
|
||||
meter.DefaultMeter.Counter(semconv.PoolPutTotal, "capacity", strconv.Itoa(st.Cap())).Set(stats.Put)
|
||||
meter.DefaultMeter.Counter(semconv.PoolMisTotal, "capacity", strconv.Itoa(st.Cap())).Set(stats.Mis)
|
||||
meter.DefaultMeter.Counter(semconv.PoolRetTotal, "capacity", strconv.Itoa(st.Cap())).Set(stats.Ret)
|
||||
}
|
||||
poolsMu.Unlock()
|
||||
for range ticker.C {
|
||||
poolsMu.Lock()
|
||||
for _, st := range pools {
|
||||
stats := st.Stats()
|
||||
meter.DefaultMeter.Counter(semconv.PoolGetTotal, "capacity", strconv.Itoa(st.Cap())).Set(stats.Get)
|
||||
meter.DefaultMeter.Counter(semconv.PoolPutTotal, "capacity", strconv.Itoa(st.Cap())).Set(stats.Put)
|
||||
meter.DefaultMeter.Counter(semconv.PoolMisTotal, "capacity", strconv.Itoa(st.Cap())).Set(stats.Mis)
|
||||
meter.DefaultMeter.Counter(semconv.PoolRetTotal, "capacity", strconv.Itoa(st.Cap())).Set(stats.Ret)
|
||||
}
|
||||
poolsMu.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user