Implement CSI volume staging (one device per volume on each host)

README-ru.md

@@ -50,6 +50,7 @@ Vitastor поддерживает QEMU-драйвер, протоколы NBD и
   - Параметры
     - [Общие](docs/config/common.ru.md)
     - [Сетевые](docs/config/network.ru.md)
+    - [Клиентский код](docs/config/client.en.md)
     - [Глобальные дисковые параметры](docs/config/layout-cluster.ru.md)
     - [Дисковые параметры OSD](docs/config/layout-osd.ru.md)
     - [Прочие параметры OSD](docs/config/osd.ru.md)

README.md

@@ -50,6 +50,7 @@ Read more details below in the documentation.
   - Parameter Reference
     - [Common](docs/config/common.en.md)
     - [Network](docs/config/network.en.md)
+    - [Client](docs/config/client.en.md)
     - [Global Disk Layout](docs/config/layout-cluster.en.md)
     - [OSD Disk Layout](docs/config/layout-osd.en.md)
     - [OSD Runtime Parameters](docs/config/osd.en.md)

csi/go.mod

@@ -9,7 +9,8 @@ require (
 	golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 	google.golang.org/grpc v1.33.1
-	k8s.io/klog v1.1.0
+	google.golang.org/protobuf v1.24.0
+	k8s.io/klog v1.0.0
 	k8s.io/utils v0.0.0-20210305010621-2afb4311ab10
 )
 

csi/go.sum

@@ -6,9 +6,9 @@ cloud.google.com/go v0.45.1/go.mod h1:RpBamKRgapWJb87xiFSdk4g1CME7QZg3uwTez+TSTj
 cloud.google.com/go v0.46.3/go.mod h1:a6bKKbmY7er1mI7TEI4lsAkts/mkhTSZK8w33B4RAg0=
 cloud.google.com/go v0.51.0/go.mod h1:hWtGJ6gnXH+KgDv+V0zFGDvpi07n3z8ZNj3T1RW0Gcw=
 cloud.google.com/go/bigquery v1.0.1/go.mod h1:i/xbL2UlR5RvWAURpBYZTtm/cXjCha9lbfbpx4poX+o=
-cloud.google.com/go/datastore v1.1.0/go.mod h1:LXYbyblFSglQ5pkeyhO+Qmw7ukd3C+pD7TKLgZqpHYE=
+cloud.google.com/go/datastore v1.0.0/go.mod h1:LXYbyblFSglQ5pkeyhO+Qmw7ukd3C+pD7TKLgZqpHYE=
 cloud.google.com/go/pubsub v1.0.1/go.mod h1:R0Gpsv3s54REJCy4fxDixWD93lHJMoZTyQ2kNxGRt3I=
-cloud.google.com/go/storage v1.1.0/go.mod h1:IhtSnM/ZTZV8YYJWCY8RULGVqBDmpoyjwiyrjsg+URw=
+cloud.google.com/go/storage v1.0.0/go.mod h1:IhtSnM/ZTZV8YYJWCY8RULGVqBDmpoyjwiyrjsg+URw=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78/go.mod h1:LmzpDX56iTiv29bbRTIsUNlaFfuhWRQBWjQdVyAevI8=
 github.com/Azure/go-autorest/autorest v0.9.0/go.mod h1:xyHB1BMZT0cuDHU7I0+g046+BFDTQ8rEZB0s4Yfa6bI=
@@ -25,14 +25,14 @@ github.com/Azure/go-autorest/tracing v0.5.0/go.mod h1:r/s2XiOKccPW3HrqB+W0TQzfbt
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
 github.com/NYTimes/gziphandler v0.0.0-20170623195520-56545f4a5d46/go.mod h1:3wb06e3pkSAbeQ52E9H9iFoQsEEwGN64994WTCIhntQ=
-github.com/PuerkitoBio/purell v1.1.0/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0=
+github.com/PuerkitoBio/purell v1.0.0/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0=
 github.com/PuerkitoBio/urlesc v0.0.0-20160726150825-5bd2802263f2/go.mod h1:uGdkoq3SwY9Y+13GIhn11/XLaGBb4BfwItxLd5jeuXE=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
-github.com/beorn7/perks v1.1.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
+github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/blang/semver v3.5.0+incompatible/go.mod h1:kRBLl5iJ+tD4TcOOxsy/0fnwebNt5EWlYSAyrTnjyyk=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
@@ -92,13 +92,13 @@ github.com/golang/protobuf v1.4.1/go.mod h1:U8fpvMrcmy5pZrNK1lt4xCsGvpyWQ/VVv6QD
 github.com/golang/protobuf v1.4.2 h1:+Z5KGCizgyZCbGh1KZqA0fcLLkwbsjIzS4aV2v7wJX0=
 github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
-github.com/google/btree v1.1.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
+github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/google/gofuzz v1.1.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/gofuzz v1.1.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
 github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
@@ -112,7 +112,7 @@ github.com/googleapis/gnostic v0.4.1/go.mod h1:LRhVm6pbyptWbWbuZ38d1eyptfvIytN3i
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
-github.com/hpcloud/tail v1.1.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
+github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
 github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/imdario/mergo v0.3.5/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
@@ -121,7 +121,7 @@ github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1
 github.com/jstemmer/go-junit-report v0.9.1/go.mod h1:Brl9GWCQeLvo8nXZwPNNblvFj/XSXhF0NWZEnDohbsk=
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
 github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
-github.com/kisielk/gotool v1.1.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
+github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
@@ -153,10 +153,10 @@ github.com/peterbourgon/diskv v2.0.1+incompatible/go.mod h1:uqqh8zWWbv1HBMNONnaR
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
-github.com/pmezard/go-difflib v1.1.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
-github.com/pmezard/go-difflib v1.1.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
-github.com/prometheus/client_golang v1.1.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
+github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
 github.com/prometheus/client_golang v1.7.1/go.mod h1:PY5Wy2awLA44sXw4AOSfFBetzPP4j5+D6mVACh+pe2M=
 github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
 github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
@@ -326,13 +326,13 @@ google.golang.org/protobuf v1.24.0 h1:UhZDfRO8JRQru4/+LlLE0BRKGF8L+PICnvYZmx/fEG
 google.golang.org/protobuf v1.24.0/go.mod h1:r/3tXBNzIEhYS9I1OUVjXDlt8tc493IdKGjtUeSXeh4=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.1.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.1.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
-gopkg.in/check.v1 v1.1.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
 gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
-gopkg.in/tomb.v1 v1.1.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
+gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
@@ -351,8 +351,8 @@ k8s.io/apimachinery v0.19.0/go.mod h1:DnPGDnARWFvYa3pMHgSxtbZb7gpzzAZ1pTfaUNDVlm
 k8s.io/client-go v0.19.0/go.mod h1:H9E/VT95blcFQnlyShFgnFT9ZnJOAceiUHM3MlRC+mU=
 k8s.io/component-base v0.19.0/go.mod h1:dKsY8BxkA+9dZIAh2aWJLL/UdASFDNtGYTCItL4LM7Y=
 k8s.io/gengo v0.0.0-20200413195148-3a45101e95ac/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8IAqLxYwwyPxAX1Pzy0ii0=
-k8s.io/klog v1.1.0 h1:Pt+yjF5aB1xDSVbau4VsWe+dQNzA0qv1LlXdC2dF6Q8=
-k8s.io/klog v1.1.0/go.mod h1:4Bi6QPql/J/LkTDqv7R/cd3hPo4k2DG6Ptcz060Ez5I=
+k8s.io/klog v1.0.0 h1:Pt+yjF5aB1xDSVbau4VsWe+dQNzA0qv1LlXdC2dF6Q8=
+k8s.io/klog v1.0.0/go.mod h1:4Bi6QPql/J/LkTDqv7R/cd3hPo4k2DG6Ptcz060Ez5I=
 k8s.io/klog/v2 v2.0.0/go.mod h1:PBfzABfn139FHAV07az/IF9Wp1bkk3vpT2XSJ76fSDE=
 k8s.io/klog/v2 v2.2.0 h1:XRvcwJozkgZ1UQJmfMGpvRthQHOvihEhYtDfAaxMz/A=
 k8s.io/klog/v2 v2.2.0/go.mod h1:Od+F08eJP+W3HUb4pSrPpgp9DGU4GzlpG/TmITuYh/Y=

csi/src/controllerserver.go

@@ -20,6 +20,7 @@ import (
 
     "google.golang.org/grpc/codes"
     "google.golang.org/grpc/status"
+    "google.golang.org/protobuf/types/known/timestamppb"
 
     "github.com/container-storage-interface/spec/lib/go/csi"
 )
@@ -45,6 +46,7 @@ type InodeConfig struct
     ParentPool uint64 `json:"parent_pool,omitempty"`
     ParentId uint64 `json:"parent_id,omitempty"`
     Readonly bool `json:"readonly,omitempty"`
+    CreateTs uint64 `json:"create_ts,omitempty"`
 }
 
 type ControllerServer struct
@@ -184,21 +186,11 @@ func (cs *ControllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
     {
         if (strings.Index(err.Error(), "already exists") > 0)
         {
-            stat, err := invokeCLI(ctxVars, []string{ "ls", "--json", volName })
+            inodeCfg, err := invokeList(ctxVars, volName, true)
             if (err != nil)
             {
                 return nil, err
             }
-            var inodeCfg []InodeConfig
-            err = json.Unmarshal(stat, &inodeCfg)
-            if (err != nil)
-            {
-                return nil, status.Error(codes.Internal, "Invalid JSON in vitastor-cli ls: "+err.Error())
-            }
-            if (len(inodeCfg) == 0)
-            {
-                return nil, status.Error(codes.Internal, "vitastor-cli create said that image already exists, but ls can't find it")
-            }
             if (inodeCfg[0].Size < uint64(volSize))
             {
                 return nil, status.Error(codes.Internal, "image "+volName+" is already created, but size is less than expected")
@@ -230,15 +222,15 @@ func (cs *ControllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol
         return nil, status.Error(codes.InvalidArgument, "request cannot be empty")
     }
 
-    ctxVars := make(map[string]string)
-    err := json.Unmarshal([]byte(req.VolumeId), &ctxVars)
+    volVars := make(map[string]string)
+    err := json.Unmarshal([]byte(req.VolumeId), &volVars)
     if (err != nil)
     {
         return nil, status.Error(codes.Internal, "volume ID not in JSON format")
     }
-    volName := ctxVars["name"]
+    volName := volVars["name"]
 
-    ctxVars, _, _ = GetConnectionParams(ctxVars)
+    ctxVars, _, _ := GetConnectionParams(volVars)
 
     _, err = invokeCLI(ctxVars, []string{ "rm", volName })
     if (err != nil)
@@ -344,6 +336,8 @@ func (cs *ControllerServer) ControllerGetCapabilities(ctx context.Context, req *
         csi.ControllerServiceCapability_RPC_LIST_VOLUMES,
         csi.ControllerServiceCapability_RPC_EXPAND_VOLUME,
         csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
+        csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS,
+        // TODO: csi.ControllerServiceCapability_RPC_CLONE_VOLUME,
     } {
         controllerServerCapabilities = append(controllerServerCapabilities, functionControllerServerCapabilities(capability))
     }
@@ -353,28 +347,219 @@ func (cs *ControllerServer) ControllerGetCapabilities(ctx context.Context, req *
     }, nil
 }
 
+func invokeList(ctxVars map[string]string, pattern string, expectExist bool) ([]InodeConfig, error)
+{
+    stat, err := invokeCLI(ctxVars, []string{ "ls", "--json", pattern })
+    if (err != nil)
+    {
+        return nil, err
+    }
+    var inodeCfg []InodeConfig
+    err = json.Unmarshal(stat, &inodeCfg)
+    if (err != nil)
+    {
+        return nil, status.Error(codes.Internal, "Invalid JSON in vitastor-cli ls: "+err.Error())
+    }
+    if (expectExist && len(inodeCfg) == 0)
+    {
+        return nil, status.Error(codes.Internal, "Can't find expected image "+pattern+" via vitastor-cli ls")
+    }
+    return inodeCfg, nil
+}
+
 // CreateSnapshot create snapshot of an existing PV
 func (cs *ControllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateSnapshotRequest) (*csi.CreateSnapshotResponse, error)
 {
-    return nil, status.Error(codes.Unimplemented, "")
+    klog.Infof("received controller create snapshot request %+v", protosanitizer.StripSecrets(req))
+    if (req == nil)
+    {
+        return nil, status.Errorf(codes.InvalidArgument, "request cannot be empty")
+    }
+    if (req.SourceVolumeId == "" || req.Name == "")
+    {
+        return nil, status.Error(codes.InvalidArgument, "source volume ID and snapshot name are required fields")
+    }
+
+    // snapshot name
+    snapName := req.Name
+
+    // req.VolumeId is an ugly json string in our case :)
+    ctxVars := make(map[string]string)
+    err := json.Unmarshal([]byte(req.SourceVolumeId), &ctxVars)
+    if (err != nil)
+    {
+        return nil, status.Error(codes.Internal, "volume ID not in JSON format")
+    }
+    volName := ctxVars["name"]
+
+    // Create image using vitastor-cli
+    _, err = invokeCLI(ctxVars, []string{ "create", "--snapshot", snapName, volName })
+    if (err != nil && strings.Index(err.Error(), "already exists") <= 0)
+    {
+        return nil, err
+    }
+
+    // Check created snapshot
+    inodeCfg, err := invokeList(ctxVars, volName+"@"+snapName, true)
+    if (err != nil)
+    {
+        return nil, err
+    }
+
+    // Use ugly JSON snapshot ID again, DeleteSnapshot doesn't have context :-(
+    ctxVars["snapshot"] = snapName
+    snapIdJson, _ := json.Marshal(ctxVars)
+    return &csi.CreateSnapshotResponse{
+        Snapshot: &csi.Snapshot{
+            SizeBytes: int64(inodeCfg[0].Size),
+            SnapshotId: string(snapIdJson),
+            SourceVolumeId: req.SourceVolumeId,
+            CreationTime: &timestamppb.Timestamp{ Seconds: int64(inodeCfg[0].CreateTs) },
+            ReadyToUse: true,
+        },
+    }, nil
 }
 
 // DeleteSnapshot delete provided snapshot of a PV
 func (cs *ControllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteSnapshotRequest) (*csi.DeleteSnapshotResponse, error)
 {
-    return nil, status.Error(codes.Unimplemented, "")
+    klog.Infof("received controller delete snapshot request %+v", protosanitizer.StripSecrets(req))
+    if (req == nil)
+    {
+        return nil, status.Errorf(codes.InvalidArgument, "request cannot be empty")
+    }
+    if (req.SnapshotId == "")
+    {
+        return nil, status.Error(codes.InvalidArgument, "snapshot ID is a required field")
+    }
+
+    volVars := make(map[string]string)
+    err := json.Unmarshal([]byte(req.SnapshotId), &volVars)
+    if (err != nil)
+    {
+        return nil, status.Error(codes.Internal, "snapshot ID not in JSON format")
+    }
+    volName := volVars["name"]
+    snapName := volVars["snapshot"]
+
+    ctxVars, _, _ := GetConnectionParams(volVars)
+
+    _, err = invokeCLI(ctxVars, []string{ "rm", volName+"@"+snapName })
+    if (err != nil)
+    {
+        return nil, err
+    }
+
+    return &csi.DeleteSnapshotResponse{}, nil
 }
 
 // ListSnapshots list the snapshots of a PV
 func (cs *ControllerServer) ListSnapshots(ctx context.Context, req *csi.ListSnapshotsRequest) (*csi.ListSnapshotsResponse, error)
 {
-    return nil, status.Error(codes.Unimplemented, "")
+    klog.Infof("received controller list snapshots request %+v", protosanitizer.StripSecrets(req))
+    if (req == nil)
+    {
+        return nil, status.Error(codes.InvalidArgument, "request cannot be empty")
+    }
+
+    volVars := make(map[string]string)
+    err := json.Unmarshal([]byte(req.SourceVolumeId), &volVars)
+    if (err != nil)
+    {
+        return nil, status.Error(codes.Internal, "volume ID not in JSON format")
+    }
+    volName := volVars["name"]
+    ctxVars, _, _ := GetConnectionParams(volVars)
+
+    inodeCfg, err := invokeList(ctxVars, volName+"@*", false)
+    if (err != nil)
+    {
+        return nil, err
+    }
+
+    resp := &csi.ListSnapshotsResponse{}
+    for _, ino := range inodeCfg
+    {
+        snapName := ino.Name[len(volName)+1:]
+        if (len(req.StartingToken) > 0 && snapName < req.StartingToken)
+        {
+        }
+        else if (req.MaxEntries == 0 || len(resp.Entries) < int(req.MaxEntries))
+        {
+            volVars["snapshot"] = snapName
+            snapIdJson, _ := json.Marshal(volVars)
+            resp.Entries = append(resp.Entries, &csi.ListSnapshotsResponse_Entry{
+                Snapshot: &csi.Snapshot{
+                    SizeBytes: int64(ino.Size),
+                    SnapshotId: string(snapIdJson),
+                    SourceVolumeId: req.SourceVolumeId,
+                    CreationTime: &timestamppb.Timestamp{ Seconds: int64(ino.CreateTs) },
+                    ReadyToUse: true,
+                },
+            })
+        }
+        else
+        {
+            resp.NextToken = snapName
+            break
+        }
+    }
+
+    return resp, nil
 }
 
-// ControllerExpandVolume resizes a volume
+// ControllerExpandVolume increases the size of a volume
 func (cs *ControllerServer) ControllerExpandVolume(ctx context.Context, req *csi.ControllerExpandVolumeRequest) (*csi.ControllerExpandVolumeResponse, error)
 {
-    return nil, status.Error(codes.Unimplemented, "")
+    klog.Infof("received controller expand volume request %+v", protosanitizer.StripSecrets(req))
+    if (req == nil)
+    {
+        return nil, status.Error(codes.InvalidArgument, "request cannot be empty")
+    }
+    if (req.VolumeId == "" || req.CapacityRange == nil || req.CapacityRange.RequiredBytes == 0)
+    {
+        return nil, status.Error(codes.InvalidArgument, "VolumeId, CapacityRange and RequiredBytes are required fields")
+    }
+
+    volVars := make(map[string]string)
+    err := json.Unmarshal([]byte(req.VolumeId), &volVars)
+    if (err != nil)
+    {
+        return nil, status.Error(codes.Internal, "volume ID not in JSON format")
+    }
+    volName := volVars["name"]
+    ctxVars, _, _ := GetConnectionParams(volVars)
+
+    inodeCfg, err := invokeList(ctxVars, volName, true)
+    if (err != nil)
+    {
+        return nil, err
+    }
+
+    if (req.CapacityRange.RequiredBytes > 0 && inodeCfg[0].Size < uint64(req.CapacityRange.RequiredBytes))
+    {
+        sz := ((req.CapacityRange.RequiredBytes+4095)/4096)*4096
+        _, err := invokeCLI(ctxVars, []string{ "modify", "--inc_size", "1", "--resize", fmt.Sprintf("%d", sz), volName })
+        if (err != nil)
+        {
+            return nil, err
+        }
+        inodeCfg, err := invokeList(ctxVars, volName, true)
+        if (err != nil)
+        {
+            return nil, err
+        }
+        return &csi.ControllerExpandVolumeResponse{
+            CapacityBytes: int64(inodeCfg[0].Size),
+            // Node expansion is required for FS
+            NodeExpansionRequired: req.VolumeCapability == nil || req.VolumeCapability.GetBlock() == nil,
+        }, nil
+    }
+
+    return &csi.ControllerExpandVolumeResponse{
+        CapacityBytes: int64(inodeCfg[0].Size),
+        NodeExpansionRequired: false,
+    }, nil
 }
 
 // ControllerGetVolume get volume info

csi/src/identityserver.go

@@ -49,6 +49,13 @@ func (is *IdentityServer) GetPluginCapabilities(ctx context.Context, req *csi.Ge
                     },
                 },
             },
+            {
+                Type: &csi.PluginCapability_VolumeExpansion_{
+                    VolumeExpansion: &csi.PluginCapability_VolumeExpansion{
+                        Type: csi.PluginCapability_VolumeExpansion_OFFLINE,
+                    },
+                },
+            },
         },
     }, nil
 }

csi/src/nodeserver.go

@@ -37,43 +37,13 @@ func NewNodeServer(driver *Driver) *NodeServer
     }
 }
 
-// NodeStageVolume mounts the volume to a staging path on the node.
-func (ns *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error)
+func (ns *NodeServer) checkMountPoint(targetPath string, isBlock bool)
 {
-    return &csi.NodeStageVolumeResponse{}, nil
-}
-
-// NodeUnstageVolume unstages the volume from the staging path
-func (ns *NodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstageVolumeRequest) (*csi.NodeUnstageVolumeResponse, error)
-{
-    return &csi.NodeUnstageVolumeResponse{}, nil
-}
-
-func Contains(list []string, s string) bool
-{
-    for i := 0; i < len(list); i++
-    {
-        if (list[i] == s)
-        {
-            return true
-        }
-    }
-    return false
-}
-
-// NodePublishVolume mounts the volume mounted to the staging path to the target path
-func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error)
-{
-    klog.Infof("received node publish volume request %+v", protosanitizer.StripSecrets(req))
-
-    targetPath := req.GetTargetPath()
-    isBlock := req.GetVolumeCapability().GetBlock() != nil
-
     // Check that it's not already mounted
-    _, error := mount.IsNotMountPoint(ns.mounter, targetPath)
-    if (error != nil)
+    _, err := mount.IsNotMountPoint(ns.mounter, targetPath)
+    if (err != nil)
     {
-        if (os.IsNotExist(error))
+        if (os.IsNotExist(err))
         {
             if (isBlock)
             {
@@ -81,13 +51,13 @@ func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis
                 if (err != nil)
                 {
                     klog.Errorf("failed to create block device mount target %s with error: %v", targetPath, err)
-                    return nil, status.Error(codes.Internal, err.Error())
+                    return status.Error(codes.Internal, err.Error("failed to create block device mount target"))
                 }
                 err = pathFile.Close()
                 if (err != nil)
                 {
                     klog.Errorf("failed to close %s with error: %v", targetPath, err)
-                    return nil, status.Error(codes.Internal, err.Error())
+                    return status.Error(codes.Internal, err.Error("failed to create block device mount target"))
                 }
             }
             else
@@ -96,15 +66,32 @@ func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis
                 if (err != nil)
                 {
                     klog.Errorf("failed to create fs mount target %s with error: %v", targetPath, err)
-                    return nil, status.Error(codes.Internal, err.Error())
+                    return status.Error(codes.Internal, err.Error("failed to create fs mount target"))
                 }
             }
         }
         else
         {
-            return nil, status.Error(codes.Internal, error.Error())
+            return status.Error(codes.Internal, err)
         }
     }
+    return nil
+}
+
+// NodeStageVolume mounts the volume to a staging path on the node.
+func (ns *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error)
+{
+    klog.Infof("received node stage volume request %+v", protosanitizer.StripSecrets(req))
+
+    targetPath := req.GetStagingTargetPath()
+    isBlock := req.GetVolumeCapability().GetBlock() != nil
+
+    // Check that it's not already mounted
+    err := ns.checkMountPoint(targetPath, isBlock)
+    if (err != nil)
+    {
+        return nil, err
+    }
 
     ctxVars := make(map[string]string)
     err := json.Unmarshal([]byte(req.VolumeId), &ctxVars)
@@ -121,7 +108,6 @@ func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis
     }
 
     // Map NBD device
-    // FIXME: Check if already mapped
     args := []string{
         "map", "--etcd_address", strings.Join(etcdUrl, ","),
         "--etcd_prefix", etcdPrefix,
@@ -226,6 +212,60 @@ func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis
         }
         return nil, status.Error(codes.Internal, err.Error())
     }
+
+    return &csi.NodeStageVolumeResponse{}, nil
+}
+
+// NodeUnstageVolume unstages the volume from the staging path
+func (ns *NodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstageVolumeRequest) (*csi.NodeUnstageVolumeResponse, error)
+{
+    klog.Infof("received node unstage volume request %+v", protosanitizer.StripSecrets(req))
+    err := ns.unmount(req.GetStagingTargetPath())
+    if (err != nil)
+    {
+        return nil, err
+    }
+    return &csi.NodeUnstageVolumeResponse{}, nil
+}
+
+func Contains(list []string, s string) bool
+{
+    for i := 0; i < len(list); i++
+    {
+        if (list[i] == s)
+        {
+            return true
+        }
+    }
+    return false
+}
+
+// NodePublishVolume mounts the volume mounted to the staging path to the target path
+func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error)
+{
+    klog.Infof("received node publish volume request %+v", protosanitizer.StripSecrets(req))
+
+    stagingTargetPath := req.GetStagingTargetPath()
+    targetPath := req.GetTargetPath()
+    isBlock := req.GetVolumeCapability().GetBlock() != nil
+
+    err := ns.checkMountPoint(targetPath, isBlock)
+    if (err != nil)
+    {
+        return nil, err
+    }
+
+    diskMounter := &mount.SafeFormatAndMount{Interface: ns.mounter, Exec: utilexec.New()}
+    err = diskMounter.Mount(stagingTargetPath, targetPath, "", []string{"bind"})
+    if (err != nil)
+    {
+        klog.Errorf(
+            "failed to bind staging path (%s) to path (%s) for volume (%s) error: %s",
+            stagingPath, targetPath, volName, err,
+        )
+        return nil, status.Error(codes.Internal, err.Error("failed to bind staging path"))
+    }
+
     return &csi.NodePublishVolumeResponse{}, nil
 }
 
@@ -233,25 +273,34 @@ func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis
 func (ns *NodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpublishVolumeRequest) (*csi.NodeUnpublishVolumeResponse, error)
 {
     klog.Infof("received node unpublish volume request %+v", protosanitizer.StripSecrets(req))
-    targetPath := req.GetTargetPath()
+    err := ns.unmount(req.GetTargetPath())
+    if (err != nil)
+    {
+        return nil, err
+    }
+    return &csi.NodeUnpublishVolumeResponse{}, nil
+}
+
+func (ns *NodeServer) unmount(targetPath string) error
+{
     devicePath, refCount, err := mount.GetDeviceNameFromMount(ns.mounter, targetPath)
     if (err != nil)
     {
         if (os.IsNotExist(err))
         {
-            return nil, status.Error(codes.NotFound, "Target path not found")
+            return status.Error(codes.NotFound, "Target path not found")
         }
-        return nil, status.Error(codes.Internal, err.Error())
+        return status.Error(codes.Internal, err.Error())
     }
     if (devicePath == "")
     {
-        return nil, status.Error(codes.NotFound, "Volume not mounted")
+        return status.Error(codes.NotFound, "Volume not mounted")
     }
     // unmount
     err = mount.CleanupMountPoint(targetPath, ns.mounter, false)
     if (err != nil)
     {
-        return nil, status.Error(codes.Internal, err.Error())
+        return status.Error(codes.Internal, err.Error())
     }
     // unmap NBD device
     if (refCount == 1)
@@ -262,7 +311,7 @@ func (ns *NodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpu
             klog.Errorf("failed to unmap NBD device %s: %s, error: %v", devicePath, unmapOut, unmapErr)
         }
     }
-    return &csi.NodeUnpublishVolumeResponse{}, nil
+    return nil
 }
 
 // NodeGetVolumeStats returns volume capacity statistics available for the volume

docs/config/client.ru.md

@@ -67,8 +67,8 @@
 явно не разрешена клиентом, т.к. если клиент не знает, что запросы записи
 буферизуются, это может приводить к потере данных. Поэтому в старых версиях
 клиентских драйверов буферизация записи не включается вообще, в новых
-версиях QEMU-драйвера включается только если разрешена опцией диска
-`-blockdev cache.direct=false`, а в fio - только если нет опция `-direct=1`.
+версиях QEMU-драйвера включается, только если разрешена опцией диска
+`-blockdev cache.direct=false`, а в fio - только если нет опции `-direct=1`.
 В NBD и NFS драйверах буферизация записи разрешена по умолчанию.
 
 Можно обойти и это ограничение с помощью параметра `client_writeback_allowed`,

docs/config/src/client.yml

@@ -87,8 +87,8 @@
     явно не разрешена клиентом, т.к. если клиент не знает, что запросы записи
     буферизуются, это может приводить к потере данных. Поэтому в старых версиях
     клиентских драйверов буферизация записи не включается вообще, в новых
-    версиях QEMU-драйвера включается только если разрешена опцией диска
-    `-blockdev cache.direct=false`, а в fio - только если нет опция `-direct=1`.
+    версиях QEMU-драйвера включается, только если разрешена опцией диска
+    `-blockdev cache.direct=false`, а в fio - только если нет опции `-direct=1`.
     В NBD и NFS драйверах буферизация записи разрешена по умолчанию.
 
     Можно обойти и это ограничение с помощью параметра `client_writeback_allowed`,

docs/intro/features.en.md

@@ -51,13 +51,15 @@
 
 The following features are planned for the future:
 
+- File system
+- Control plane optimisation
 - Other administrative tools
 - Web GUI
 - OpenNebula plugin
-- iSCSI proxy
+- iSCSI and NVMeoF gateways
 - Multi-threaded client
 - Faster failover
+- S3
 - Tiered storage (SSD caching)
 - NVDIMM support
 - Compression (possibly)
-- Read caching using system page cache (possibly)

docs/intro/features.ru.md

@@ -51,12 +51,15 @@
 
 ## Планы развития
 
+- Файловая система
+- Оптимизация слоя управления
 - Другие инструменты администрирования
 - Web-интерфейс
 - Плагин для OpenNebula
-- iSCSI-прокси
+- iSCSI и NVMeoF прокси
 - Многопоточный клиент
 - Более быстрое переключение при отказах
+- S3
 - Поддержка SSD-кэширования (tiered storage)
 - Поддержка NVDIMM
 - Возможно, сжатие

mon/90-vitastor.rules

@@ -3,5 +3,5 @@ SUBSYSTEM=="block", ENV{ID_PART_ENTRY_TYPE}=="e7009fac-a5a1-4d72-af72-53de130599
     IMPORT{program}="/usr/bin/vitastor-disk udev $devnode", \
     SYMLINK+="vitastor/$env{VITASTOR_ALIAS}"
 
-ENV{VITASTOR_OSD_NUM}!="", ACTION=="add", RUN{program}+="/usr/bin/systemctl enable --now vitastor-osd@$env{VITASTOR_OSD_NUM}"
-ENV{VITASTOR_OSD_NUM}!="", ACTION=="remove", RUN{program}+="/usr/bin/systemctl disable --now vitastor-osd@$env{VITASTOR_OSD_NUM}"
+ENV{VITASTOR_OSD_NUM}!="", ACTION=="add", RUN{program}+="/usr/bin/systemctl enable --now --no-block vitastor-osd@$env{VITASTOR_OSD_NUM}"
+ENV{VITASTOR_OSD_NUM}!="", ACTION=="remove", RUN{program}+="/usr/bin/systemctl disable --now --no-block vitastor-osd@$env{VITASTOR_OSD_NUM}"

mon/mon.js

@@ -397,8 +397,8 @@ class Mon
         this.etcd_prefix = this.etcd_prefix.replace(/\/\/+/g, '/').replace(/^\/?(.*[^\/])\/?$/, '/$1');
         this.etcd_start_timeout = (config.etcd_start_timeout || 5) * 1000;
         this.state = JSON.parse(JSON.stringify(this.constructor.etcd_tree));
+        this.prev_stats = { osd_stats: {}, osd_diff: {} };
         this.signals_set = false;
-        this.stat_time = Date.now();
         this.ws = null;
         this.ws_alive = false;
         this.ws_keepalive_timer = null;
@@ -552,9 +552,9 @@ class Mon
             const cur_addr = this.pick_next_etcd();
             const base = 'ws'+cur_addr.substr(4);
             let now = Date.now();
-            if (tried[base] && now-tried[base] < timeout)
+            if (tried[base] && now-tried[base] < this.etcd_start_timeout)
             {
-                await new Promise(ok => setTimeout(ok, timeout-(now-tried[base])));
+                await new Promise(ok => setTimeout(ok, this.etcd_start_timeout-(now-tried[base])));
                 now = Date.now();
             }
             tried[base] = now;
@@ -1459,15 +1459,15 @@ class Mon
         }
     }
 
-    derive_osd_stats(st, prev)
+    derive_osd_stats(st, prev, prev_diff)
     {
         const zero_stats = { op: { bps: 0n, iops: 0n, lat: 0n }, subop: { iops: 0n, lat: 0n }, recovery: { bps: 0n, iops: 0n } };
-        const diff = { op_stats: {}, subop_stats: {}, recovery_stats: {} };
-        if (!st || !st.time || prev && (prev.time || this.stat_time/1000) >= st.time)
+        const diff = { op_stats: {}, subop_stats: {}, recovery_stats: {}, inode_stats: {} };
+        if (!st || !st.time || !prev || prev.time >= st.time)
         {
-            return diff;
+            return prev_diff || diff;
         }
-        const timediff = BigInt(st.time*1000 - (prev && prev.time*1000 || this.stat_time));
+        const timediff = BigInt(st.time*1000 - prev.time*1000);
         for (const op in st.op_stats||{})
         {
             const pr = prev && prev.op_stats && prev.op_stats[op];
@@ -1499,25 +1499,47 @@ class Mon
             if (n > 0)
                 diff.recovery_stats[op] = { ...c, bps: b*1000n/timediff, iops: n*1000n/timediff };
         }
+        for (const pool_id in st.inode_stats||{})
+        {
+            const pool_diff = diff.inode_stats[pool_id] = {};
+            for (const inode_num in st.inode_stats[pool_id])
+            {
+                const inode_diff = diff.inode_stats[pool_id][inode_num] = {};
+                for (const op of [ 'read', 'write', 'delete' ])
+                {
+                    const c = st.inode_stats[pool_id][inode_num][op];
+                    const pr = prev && prev.inode_stats && prev.inode_stats[pool_id] &&
+                        prev.inode_stats[pool_id][inode_num] && prev.inode_stats[pool_id][inode_num][op];
+                    const n = BigInt(c.count||0) - BigInt(pr && pr.count||0);
+                    inode_diff[op] = {
+                        bps: (BigInt(c.bytes||0) - BigInt(pr && pr.bytes||0))*1000n/timediff,
+                        iops: n*1000n/timediff,
+                        lat: (BigInt(c.usec||0) - BigInt(pr && pr.usec||0))/(n || 1n),
+                    };
+                }
+            }
+        }
         return diff;
     }
 
-    sum_op_stats(timestamp, prev_stats)
+    sum_op_stats()
     {
-        const sum_diff = { op_stats: {}, subop_stats: {}, recovery_stats: {} };
-        if (!prev_stats || prev_stats.timestamp >= timestamp)
+        for (const osd in this.state.osd.stats)
         {
-            return sum_diff;
+            const cur = { ...this.state.osd.stats[osd], inode_stats: this.state.osd.inodestats[osd]||{} };
+            this.prev_stats.osd_diff[osd] = this.derive_osd_stats(
+                cur, this.prev_stats.osd_stats[osd], this.prev_stats.osd_diff[osd]
+            );
+            this.prev_stats.osd_stats[osd] = cur;
         }
-        const tm = BigInt(timestamp - (prev_stats.timestamp || 0));
+        const sum_diff = { op_stats: {}, subop_stats: {}, recovery_stats: {} };
         // Sum derived values instead of deriving summed
         for (const osd in this.state.osd.stats)
         {
-            const derived = this.derive_osd_stats(this.state.osd.stats[osd],
-                this.prev_stats && this.prev_stats.osd_stats && this.prev_stats.osd_stats[osd]);
-            for (const type in derived)
+            const derived = this.prev_stats.osd_diff[osd];
+            for (const type in sum_diff)
             {
-                for (const op in derived[type])
+                for (const op in derived[type]||{})
                 {
                     for (const k in derived[type][op])
                     {
@@ -1574,14 +1596,14 @@ class Mon
         return { object_counts, object_bytes };
     }
 
-    sum_inode_stats(prev_stats, timestamp, prev_timestamp)
+    sum_inode_stats()
     {
         const inode_stats = {};
         const inode_stub = () => ({
             raw_used: 0n,
-            read: { count: 0n, usec: 0n, bytes: 0n },
-            write: { count: 0n, usec: 0n, bytes: 0n },
-            delete: { count: 0n, usec: 0n, bytes: 0n },
+            read: { count: 0n, usec: 0n, bytes: 0n, bps: 0n, iops: 0n, lat: 0n },
+            write: { count: 0n, usec: 0n, bytes: 0n, bps: 0n, iops: 0n, lat: 0n },
+            delete: { count: 0n, usec: 0n, bytes: 0n, bps: 0n, iops: 0n, lat: 0n },
         });
         const seen_pools = {};
         for (const pool_id in this.state.config.pools)
@@ -1633,11 +1655,25 @@ class Mon
                 }
             }
         }
-        if (prev_stats && prev_timestamp >= timestamp)
+        for (const osd in this.prev_stats.osd_diff)
         {
-            prev_stats = null;
+            for (const pool_id in this.prev_stats.osd_diff[osd].inode_stats)
+            {
+                for (const inode_num in this.prev_stats.osd_diff[osd].inode_stats[pool_id])
+                {
+                    inode_stats[pool_id][inode_num] = inode_stats[pool_id][inode_num] || inode_stub();
+                    for (const op of [ 'read', 'write', 'delete' ])
+                    {
+                        const op_diff = this.prev_stats.osd_diff[osd].inode_stats[pool_id][inode_num][op] || {};
+                        const op_st = inode_stats[pool_id][inode_num][op];
+                        op_st.bps += op_diff.bps;
+                        op_st.iops += op_diff.iops;
+                        op_st.lat += op_diff.lat;
+                        op_st.n_osd = (op_st.n_osd || 0) + 1;
+                    }
+                }
+            }
         }
-        const tm = prev_stats ? BigInt(timestamp - prev_timestamp) : 0;
         for (const pool_id in inode_stats)
         {
             for (const inode_num in inode_stats[pool_id])
@@ -1646,11 +1682,12 @@ class Mon
                 for (const op of [ 'read', 'write', 'delete' ])
                 {
                     const op_st = inode_stats[pool_id][inode_num][op];
-                    const prev_st = prev_stats && prev_stats[pool_id] && prev_stats[pool_id][inode_num] && prev_stats[pool_id][inode_num][op];
-                    op_st.bps = prev_st ? (op_st.bytes - prev_st.bytes) * 1000n / tm : 0;
-                    op_st.iops = prev_st ? (op_st.count - prev_st.count) * 1000n / tm : 0;
-                    op_st.lat = prev_st ? (op_st.usec - prev_st.usec) / ((op_st.count - prev_st.count) || 1n) : 0;
-                    if (op_st.bps > 0 || op_st.iops > 0 || op_st.lat > 0)
+                    if (op_st.n_osd)
+                    {
+                        op_st.lat /= BigInt(op_st.n_osd);
+                        delete op_st.n_osd;
+                    }
+                    if (op_st.bps > 0 || op_st.iops > 0)
                         nonzero = true;
                 }
                 if (!nonzero && (!this.state.config.inode[pool_id] || !this.state.config.inode[pool_id][inode_num]))
@@ -1683,15 +1720,9 @@ class Mon
     async update_total_stats()
     {
         const txn = [];
-        const timestamp = Date.now();
         const { object_counts, object_bytes } = this.sum_object_counts();
-        let stats = this.sum_op_stats(timestamp, this.prev_stats);
-        let { inode_stats, seen_pools } = this.sum_inode_stats(
-            this.prev_stats ? this.prev_stats.inode_stats : null,
-            timestamp, this.prev_stats ? this.prev_stats.timestamp : null
-        );
-        this.prev_stats = { timestamp, inode_stats, osd_stats: { ...this.state.osd.stats } };
-        this.stat_time = Date.now();
+        let stats = this.sum_op_stats();
+        let { inode_stats, seen_pools } = this.sum_inode_stats();
         stats.object_counts = object_counts;
         stats.object_bytes = object_bytes;
         stats = this.serialize_bigints(stats);

src/msgr_send.cpp

@@ -119,9 +119,9 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
             try_send(cl);
         }
     }
-    else if (cl->write_msg.msg_iovlen > 0 || !try_send(cl))
+    else
     {
-        if (cl->write_state == 0)
+        if ((cl->write_msg.msg_iovlen > 0 || !try_send(cl)) && (cl->write_state == 0))
         {
             cl->write_state = CL_WRITE_READY;
             write_ready_clients.push_back(cur_op->peer_fd);