diff --git a/pkg/vgmanager/lv_attr.go b/pkg/vgmanager/lv_attr.go
new file mode 100644
index 000000000..df3af9586
--- /dev/null
+++ b/pkg/vgmanager/lv_attr.go
@@ -0,0 +1,153 @@
+package vgmanager
+
+import "fmt"
+
+// LvAttr has mapped lv_attr information, see https://linux.die.net/man/8/lvs
+// It is a complete parsing of the entire attribute byte flags that is attached to each LV.
+// This is useful when attaching logic to the state of an LV as the state of an LV can be determined
+// from the Attributes, e.g. for determining wether an LV is considered a Thin-Pool or not.
+type LvAttr struct {
+	VolumeType
+	Permissions
+	AllocationPolicy
+	Minor
+	State
+	Open
+	OpenTarget
+	Zero
+	Partial
+}
+
+func ParsedLvAttr(raw string) (LvAttr, error) {
+	if len(raw) != 10 {
+		return LvAttr{}, fmt.Errorf("%s is an invalid length lv_attr", raw)
+	}
+	return LvAttr{
+		VolumeType(raw[0]),
+		Permissions(raw[1]),
+		AllocationPolicy(raw[2]),
+		Minor(raw[3]),
+		State(raw[4]),
+		Open(raw[5]),
+		OpenTarget(raw[6]),
+		Zero(raw[7]),
+		Partial(raw[8]),
+	}, nil
+}
+
+func (l LvAttr) String() string {
+	return fmt.Sprintf(
+		"%c%c%c%c%c%c%c%c%c",
+		l.VolumeType,
+		l.Permissions,
+		l.AllocationPolicy,
+		l.Minor,
+		l.State,
+		l.Open,
+		l.OpenTarget,
+		l.Zero,
+		l.Partial,
+	)
+}
+
+type VolumeType rune
+
+const (
+	VolumeTypeMirrored                   VolumeType = 'm'
+	VolumeTypeMirroredNoInitialSync      VolumeType = 'M'
+	VolumeTypeOrigin                     VolumeType = 'o'
+	VolumeTypeOriginWithMergingSnapshot  VolumeType = 'O'
+	VolumeTypeRAID                       VolumeType = 'r'
+	VolumeTypeRAIDNoInitialSync          VolumeType = 'R'
+	VolumeTypeSnapshot                   VolumeType = 's'
+	VolumeTypeMergingSnapshot            VolumeType = 'S'
+	VolumeTypePVMove                     VolumeType = 'p'
+	VolumeTypeVirtual                    VolumeType = 'v'
+	VolumeTypeMirrorOrRAIDImage          VolumeType = 'i'
+	VolumeTypeMirrorOrRAIDImageOutOfSync VolumeType = 'I'
+	VolumeTypeMirrorLogDevice            VolumeType = 'l'
+	VolumeTypeUnderConversion            VolumeType = 'c'
+	VolumeTypeThinVolume                 VolumeType = 'V'
+	VolumeTypeThinPool                   VolumeType = 't'
+	VolumeTypeThinPoolData               VolumeType = 'T'
+	VolumeTypeThinPoolMetadata           VolumeType = 'e'
+	VolumeTypeNone                       VolumeType = '-'
+)
+
+type Permissions rune
+
+const (
+	PermissionsWriteable                             Permissions = 'w'
+	PermissionsReadOnly                              Permissions = 'r'
+	PermissionsReadOnlyActivationOfNonReadOnlyVolume Permissions = 'R'
+	PermissionsNone                                  Permissions = '-'
+)
+
+type AllocationPolicy rune
+
+const (
+	AllocationPolicyAnywhere         AllocationPolicy = 'a'
+	AllocationPolicyAnywhereLocked   AllocationPolicy = 'A'
+	AllocationPolicyContiguous       AllocationPolicy = 'c'
+	AllocationPolicyContiguousLocked AllocationPolicy = 'C'
+	AllocationPolicyInherited        AllocationPolicy = 'i'
+	AllocationPolicyInheritedLocked  AllocationPolicy = 'I'
+	AllocationPolicyCling            AllocationPolicy = 'l'
+	AllocationPolicyClingLocked      AllocationPolicy = 'L'
+	AllocationPolicyNormal           AllocationPolicy = 'n'
+	AllocationPolicyNormalLocked     AllocationPolicy = 'N'
+	AllocationPolicyNone                              = '-'
+)
+
+type Minor rune
+
+const (
+	MinorTrue  Minor = 'm'
+	MinorFalse Minor = '-'
+)
+
+type State rune
+
+const (
+	StateActive                                State = 'a'
+	StateSuspended                                   = 's'
+	StateInvalidSnapshot                             = 'I'
+	StateSuspendedSnapshot                           = 'S'
+	StateSnapshotMergeFailed                         = 'm'
+	StateSuspendedSnapshotMergeFailed                = 'M'
+	StateMappedDevicePresentWithoutTables            = 'd'
+	StateMappedDevicePresentWithInactiveTables       = 'i'
+	StateNone                                        = '-'
+)
+
+type Open rune
+
+const (
+	OpenTrue  Open = 'o'
+	OpenFalse Open = '-'
+)
+
+type OpenTarget rune
+
+const (
+	OpenTargetMirror   = 'm'
+	OpenTargetRaid     = 'r'
+	OpenTargetSnapshot = 's'
+	OpenTargetThin     = 't'
+	OpenTargetUnknown  = 'u'
+	OpenTargetVirtual  = 'v'
+)
+
+type Zero rune
+
+const (
+	ZeroTrue  Zero = 'z'
+	ZeroFalse Zero = '-'
+)
+
+type Partial rune
+
+const (
+	PartialTrue  = 'p'
+	PartialFalse = '-'
+)
diff --git a/pkg/vgmanager/lv_attr_test.go b/pkg/vgmanager/lv_attr_test.go
new file mode 100644
index 000000000..45c5cfa27
--- /dev/null
+++ b/pkg/vgmanager/lv_attr_test.go
@@ -0,0 +1,62 @@
+package vgmanager
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestParsedLvAttr(t *testing.T) {
+	type args struct {
+		raw string
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    LvAttr
+		wantErr assert.ErrorAssertionFunc
+	}{
+		{
+			"RAID Config without Initial Sync",
+			args{raw: "Rwi-a-r---"},
+			LvAttr{
+				VolumeType:       VolumeTypeRAIDNoInitialSync,
+				Permissions:      PermissionsWriteable,
+				AllocationPolicy: AllocationPolicyInherited,
+				Minor:            MinorFalse,
+				State:            StateActive,
+				Open:             OpenFalse,
+				OpenTarget:       OpenTargetRaid,
+				Zero:             ZeroFalse,
+				Partial:          PartialFalse,
+			},
+			assert.NoError,
+		},
+		{
+			"ThinPool with Zeroing",
+			args{raw: "twi-a-tz--"},
+			LvAttr{
+				VolumeType:       VolumeTypeThinPool,
+				Permissions:      PermissionsWriteable,
+				AllocationPolicy: AllocationPolicyInherited,
+				Minor:            MinorFalse,
+				State:            StateActive,
+				Open:             OpenFalse,
+				OpenTarget:       OpenTargetThin,
+				Zero:             ZeroTrue,
+				Partial:          PartialFalse,
+			},
+			assert.NoError,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := ParsedLvAttr(tt.args.raw)
+			if !tt.wantErr(t, err, fmt.Sprintf("ParsedLvAttr(%v)", tt.args.raw)) {
+				return
+			}
+			assert.Equalf(t, tt.want, got, "ParsedLvAttr(%v)", tt.args.raw)
+		})
+	}
+}
diff --git a/pkg/vgmanager/lvm.go b/pkg/vgmanager/lvm.go
index 90068ffc2..12a2426ab 100644
--- a/pkg/vgmanager/lvm.go
+++ b/pkg/vgmanager/lvm.go
@@ -67,11 +67,12 @@ type pvsOutput struct {
 type lvsOutput struct {
 	Report []struct {
 		Lv []struct {
-			Name     string `json:"lv_name"`
-			VgName   string `json:"vg_name"`
-			PoolName string `json:"pool_lv"`
-			LvAttr   string `json:"lv_attr"`
-			LvSize   string `json:"lv_size"`
+			Name            string `json:"lv_name"`
+			VgName          string `json:"vg_name"`
+			PoolName        string `json:"pool_lv"`
+			LvAttr          string `json:"lv_attr"`
+			LvSize          string `json:"lv_size"`
+			MetadataPercent string `json:"metadata_percent"`
 		} `json:"lv"`
 	} `json:"report"`
 }
diff --git a/pkg/vgmanager/vgmanager_controller.go b/pkg/vgmanager/vgmanager_controller.go
index adfbc1dd2..ba135aa6c 100644
--- a/pkg/vgmanager/vgmanager_controller.go
+++ b/pkg/vgmanager/vgmanager_controller.go
@@ -44,9 +44,10 @@ import (
 )
 
 const (
-	ControllerName    = "vg-manager"
-	DefaultChunkSize  = "128"
-	reconcileInterval = 1 * time.Minute
+	ControllerName            = "vg-manager"
+	DefaultChunkSize          = "128"
+	reconcileInterval         = 1 * time.Minute
+	metadataWarningPercentage = 95
 )
 
 var (
@@ -181,21 +182,31 @@ func (r *VGReconciler) reconcile(ctx context.Context, volumeGroup *lvmv1alpha1.L
 			}
 		}
 
-		if devicesExist {
-			r.Log.Info("all the available devices are attached to the volume group", "VGName", volumeGroup.Name)
-			if statuserr := r.setVolumeGroupReadyStatus(ctx, volumeGroup.Name); statuserr != nil {
-				r.Log.Error(statuserr, "failed to update status", "VGName", volumeGroup.Name)
+		if !devicesExist {
+			err := fmt.Errorf("no available devices found for volume group")
+			r.Log.Error(err, err.Error(), "VGName", volumeGroup.Name)
+			if statuserr := r.setVolumeGroupFailedStatus(ctx, volumeGroup.Name, err.Error()); statuserr != nil {
+				r.Log.Error(statuserr, "failed to update status", "name", volumeGroup.Name)
 				return reconcileAgain, statuserr
 			}
-		} else {
-			errMsg := "no available devices found for volume group"
-			r.Log.Error(fmt.Errorf(errMsg), errMsg, "VGName", volumeGroup.Name)
-			if statuserr := r.setVolumeGroupFailedStatus(ctx, volumeGroup.Name, errMsg); statuserr != nil {
+			return reconcileAgain, err
+		}
+
+		if err := r.validateLVs(volumeGroup); err != nil {
+			err := fmt.Errorf("error while validating logical volumes in existing volume group: %w", err)
+			r.Log.Error(err, err.Error(), "VGName", volumeGroup.Name)
+			if statuserr := r.setVolumeGroupFailedStatus(ctx, volumeGroup.Name, err.Error()); statuserr != nil {
 				r.Log.Error(statuserr, "failed to update status", "name", volumeGroup.Name)
 				return reconcileAgain, statuserr
 			}
+			return reconcileAgain, err
 		}
 
+		r.Log.Info("all the available devices are attached to the volume group", "VGName", volumeGroup.Name)
+		if statuserr := r.setVolumeGroupReadyStatus(ctx, volumeGroup.Name); statuserr != nil {
+			r.Log.Error(statuserr, "failed to update status", "VGName", volumeGroup.Name)
+			return reconcileAgain, statuserr
+		}
 		return reconcileAgain, nil
 	}
 
@@ -370,6 +381,54 @@ func (r *VGReconciler) processDelete(ctx context.Context, volumeGroup *lvmv1alph
 	return nil
 }
 
+// validateLVs verifies that all lvs that should have been created in the volume group are present and
+// in their correct state
+func (r *VGReconciler) validateLVs(volumeGroup *lvmv1alpha1.LVMVolumeGroup) error {
+	// If we don't have a ThinPool, VG Manager has no authority about the top Level LVs inside the VG, but TopoLVM
+	if volumeGroup.Spec.ThinPoolConfig == nil {
+		return nil
+	}
+
+	resp, err := GetLVSOutput(r.executor, volumeGroup.Name)
+	if err != nil {
+		return fmt.Errorf("could not get logical volumes found inside volume group, volume group content is degraded or corrupt: %w", err)
+	}
+
+	for _, report := range resp.Report {
+		for _, lv := range report.Lv {
+			if lv.Name != volumeGroup.Spec.ThinPoolConfig.Name {
+				continue
+			}
+			lvAttr, err := ParsedLvAttr(lv.LvAttr)
+			if err != nil {
+				return fmt.Errorf("could not parse lv_attr from logical volume %s: %w", lv.Name, err)
+			}
+			if lvAttr.VolumeType != VolumeTypeThinPool {
+				return fmt.Errorf("found logical volume in volume group that is not of type Thin-Pool, "+
+					"even though there is a Thin-Pool configured: %s, lv_attr: %s,"+
+					"this is most likely a corruption of the thin pool or a setup gone wrong",
+					string(lvAttr.VolumeType), lvAttr)
+			}
+
+			if lvAttr.State != StateActive {
+				return fmt.Errorf("found inactive logical volume, maybe external repairs are formed or there is another"+
+					"entity conflicting with vg-manager, cannot proceed until volume is activated again: lv_attr: %s", lvAttr)
+			}
+			metadataPercentage, err := strconv.ParseFloat(lv.MetadataPercent, 32)
+			if err != nil {
+				return fmt.Errorf("could not ensure metadata percentage of LV due to a parsing error: %w", err)
+			}
+			if metadataPercentage > metadataWarningPercentage {
+				return fmt.Errorf("metadata partition is over %v percent filled and LVM Metadata Overflows cannot be recovered"+
+					"you should manually extend the metadata_partition or you will risk data loss: metadata_percent: %v", metadataPercentage, lv.MetadataPercent)
+			}
+
+			r.Log.Info("confirmed created logical volume has correct attributes", "lv_attr", lvAttr.String())
+		}
+	}
+	return nil
+}
+
 func (r *VGReconciler) addThinPoolToVG(vgName string, config *lvmv1alpha1.ThinPoolConfig) error {
 	resp, err := GetLVSOutput(r.executor, vgName)
 	if err != nil {
diff --git a/pkg/vgmanager/vgmanager_controller_test.go b/pkg/vgmanager/vgmanager_controller_test.go
new file mode 100644
index 000000000..f2def6e24
--- /dev/null
+++ b/pkg/vgmanager/vgmanager_controller_test.go
@@ -0,0 +1,160 @@
+package vgmanager
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/go-logr/logr/testr"
+	lvmv1alpha1 "github.com/openshift/lvm-operator/api/v1alpha1"
+	"github.com/openshift/lvm-operator/pkg/internal"
+	mockExec "github.com/openshift/lvm-operator/pkg/internal/test"
+	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/strings/slices"
+)
+
+var mockLvsOutputThinPoolValid = `
+{
+      "report": [
+          {
+              "lv": [
+                  {"lv_name":"thin-pool-1", "vg_name":"vg1", "lv_attr":"twi-a-tz--", "lv_size":"26.96g", "pool_lv":"", "origin":"", "data_percent":"0.00", "metadata_percent":"10.52", "move_pv":"", "mirror_log":"", "copy_percent":"", "convert_lv":""}
+              ]
+          }
+      ]
+  }
+`
+
+var mockLvsOutputThinPoolHighMetadataUse = `
+{
+      "report": [
+          {
+              "lv": [
+                  {"lv_name":"thin-pool-1", "vg_name":"vg1", "lv_attr":"twi-a-tz--", "lv_size":"26.96g", "pool_lv":"", "origin":"", "data_percent":"0.00", "metadata_percent":"98.52", "move_pv":"", "mirror_log":"", "copy_percent":"", "convert_lv":""}
+              ]
+          }
+      ]
+  }
+`
+var mockLvsOutputThinPoolSuspended = `
+{
+      "report": [
+          {
+              "lv": [
+                  {"lv_name":"thin-pool-1", "vg_name":"vg1", "lv_attr":"twi-s-tz--", "lv_size":"26.96g", "pool_lv":"", "origin":"", "data_percent":"0.00", "metadata_percent":"10.52", "move_pv":"", "mirror_log":"", "copy_percent":"", "convert_lv":""}
+              ]
+          }
+      ]
+  }
+`
+
+var mockLvsOutputRAID = `
+{
+      "report": [
+          {
+              "lv": [
+                  {"lv_name":"thin-pool-1", "vg_name":"vg1", "lv_attr":"rwi-a-tz--", "lv_size":"26.96g", "pool_lv":"", "origin":"", "data_percent":"0.00", "metadata_percent":"10.52", "move_pv":"", "mirror_log":"", "copy_percent":"", "convert_lv":""}
+              ]
+          }
+      ]
+  }
+`
+
+func TestVGReconciler_validateLVs(t *testing.T) {
+	type fields struct {
+		executor internal.Executor
+	}
+	type args struct {
+		volumeGroup *lvmv1alpha1.LVMVolumeGroup
+	}
+
+	lvsCommandForVG1 := []string{
+		"lvs",
+		"-S",
+		"vgname=vg1",
+		"--units",
+		"g",
+		"--reportformat",
+		"json",
+	}
+
+	mockExecutorForLVSOutput := func(output string) internal.Executor {
+		return &mockExec.MockExecutor{
+			MockExecuteCommandWithOutputAsHost: func(command string, args ...string) (string, error) {
+				if !slices.Equal(args, lvsCommandForVG1) {
+					return "", fmt.Errorf("invalid args %q", args)
+				}
+				return output, nil
+			},
+		}
+	}
+
+	tests := []struct {
+		name    string
+		fields  fields
+		args    args
+		wantErr assert.ErrorAssertionFunc
+	}{
+		{
+			name: "Valid LV",
+			fields: fields{
+				executor: mockExecutorForLVSOutput(mockLvsOutputThinPoolValid),
+			},
+			args: args{volumeGroup: &lvmv1alpha1.LVMVolumeGroup{
+				ObjectMeta: metav1.ObjectMeta{Name: "vg1", Namespace: "default"},
+				Spec: lvmv1alpha1.LVMVolumeGroupSpec{ThinPoolConfig: &lvmv1alpha1.ThinPoolConfig{
+					Name: "thin-pool-1",
+				}},
+			}},
+			wantErr: assert.NoError,
+		},
+		{
+			name: "Invalid LV due to Type not being Thin Pool",
+			fields: fields{
+				executor: mockExecutorForLVSOutput(mockLvsOutputRAID),
+			},
+			args: args{volumeGroup: &lvmv1alpha1.LVMVolumeGroup{
+				ObjectMeta: metav1.ObjectMeta{Name: "vg1", Namespace: "default"},
+				Spec: lvmv1alpha1.LVMVolumeGroupSpec{ThinPoolConfig: &lvmv1alpha1.ThinPoolConfig{
+					Name: "thin-pool-1",
+				}},
+			}},
+			wantErr: assert.Error,
+		},
+		{
+			name: "Invalid LV due to high metadata percentage",
+			fields: fields{
+				executor: mockExecutorForLVSOutput(mockLvsOutputThinPoolHighMetadataUse),
+			},
+			args: args{volumeGroup: &lvmv1alpha1.LVMVolumeGroup{
+				ObjectMeta: metav1.ObjectMeta{Name: "vg1", Namespace: "default"},
+				Spec: lvmv1alpha1.LVMVolumeGroupSpec{ThinPoolConfig: &lvmv1alpha1.ThinPoolConfig{
+					Name: "thin-pool-1",
+				}},
+			}},
+			wantErr: assert.Error,
+		},
+		{
+			name: "Invalid LV due to high metadata percentage",
+			fields: fields{
+				executor: mockExecutorForLVSOutput(mockLvsOutputThinPoolSuspended),
+			},
+			args: args{volumeGroup: &lvmv1alpha1.LVMVolumeGroup{
+				ObjectMeta: metav1.ObjectMeta{Name: "vg1", Namespace: "default"},
+				Spec: lvmv1alpha1.LVMVolumeGroupSpec{ThinPoolConfig: &lvmv1alpha1.ThinPoolConfig{
+					Name: "thin-pool-1",
+				}},
+			}},
+			wantErr: assert.Error,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := &VGReconciler{
+				Log:      testr.New(t),
+				executor: tt.fields.executor,
+			}
+			tt.wantErr(t, r.validateLVs(tt.args.volumeGroup), fmt.Sprintf("validateLVs(%v)", tt.args.volumeGroup))
+		})
+	}
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 134c912b8..67a8fa7bf 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -80,6 +80,7 @@ github.com/fsnotify/fsnotify
 ## explicit; go 1.16
 github.com/go-logr/logr
 github.com/go-logr/logr/funcr
+github.com/go-logr/logr/testr
 # github.com/go-logr/zapr v1.2.4
 ## explicit; go 1.16
 github.com/go-logr/zapr